/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you under the Apache License, Version 2.0 (the
    * "License"); you may not use this file except in compliance
    * with the License.  You may obtain a copy of the License at
    *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  package org.apache.myfaces.shared.util;
  
  import java.lang.ref.WeakReference;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.LinkedHashMap;
  import java.util.Map;
  import java.util.TreeSet;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.concurrent.locks.ReentrantLock;
  
  /**
   * A LRU cache implementation based upon ConcurrentHashMap and other techniques to reduce
   * contention and synchronization overhead to utilize multiple CPU cores more effectively.
   * <p/>
   * Note that the implementation does not follow a true LRU (least-recently-used) eviction
   * strategy. Instead it strives to remove least recently used items but when the initial
   * cleanup does not remove enough items to reach the 'acceptableWaterMark' limit, it can
   * remove more items forcefully regardless of access order.
   *
   *
   * @since solr 1.4
   * @see org.apache.solr.util.ConcurrentLRUCache
   */
  public class ConcurrentLRUCache<K, V>
  {
      //private static Logger log = Logger.getLogger(ConcurrentLRUCache.class
      //        .getName());
  
      private final ConcurrentHashMap<Object, CacheEntry<K, V>> map;
      private final int upperWaterMark;
      private final int lowerWaterMark;
      private final ReentrantLock markAndSweepLock = new ReentrantLock(true);
      private boolean isCleaning = false; // not volatile... piggybacked on other volatile vars
      private final boolean newThreadForCleanup;
      private volatile boolean islive = true;
      private final Stats stats = new Stats();
      private final int acceptableWaterMark;
      private long oldestEntry = 0; // not volatile, only accessed in the cleaning method
      private final EvictionListener<K, V> evictionListener;
      private CleanupThread cleanupThread;
  
      public ConcurrentLRUCache(int upperWaterMark, final int lowerWaterMark,
              int acceptableWatermark, int initialSize, boolean runCleanupThread,
              boolean runNewThreadForCleanup,
              EvictionListener<K, V> evictionListener)
      {
          if (upperWaterMark < 1)
          {
              throw new IllegalArgumentException("upperWaterMark must be > 0");
          }
          if (lowerWaterMark >= upperWaterMark)
          {
              throw new IllegalArgumentException(
                      "lowerWaterMark must be  < upperWaterMark");
          }
          map = new ConcurrentHashMap<Object, CacheEntry<K, V>>(initialSize);
          newThreadForCleanup = runNewThreadForCleanup;
          this.upperWaterMark = upperWaterMark;
          this.lowerWaterMark = lowerWaterMark;
          this.acceptableWaterMark = acceptableWatermark;
          this.evictionListener = evictionListener;
          if (runCleanupThread)
          {
              cleanupThread = new CleanupThread(this);
              cleanupThread.start();
          }
      }
  
      public ConcurrentLRUCache(int size, int lowerWatermark)
      {
          this(size, lowerWatermark, (int) Math
                  .floor((lowerWatermark + size) / 2), (int) Math
                  .ceil(0.75 * size), false, false, null);
      }
  
      public void setAlive(boolean live)
      {
          islive = live;
     }
 
     public V get(K key)
     {
         CacheEntry<K, V> e = map.get(key);
         if (e == null)
         {
             if (islive)
             {
                 stats.missCounter.incrementAndGet();
             }
             return null;
         }
         if (islive)
         {
             e.lastAccessed = stats.accessCounter.incrementAndGet();
         }
         return e.value;
     }
 
     public V remove(K key)
     {
         CacheEntry<K, V> cacheEntry = map.remove(key);
         if (cacheEntry != null)
         {
             stats.size.decrementAndGet();
             return cacheEntry.value;
         }
         return null;
     }
 
     public V put(K key, V val)
     {
         if (val == null)
         {
             return null;
         }
         CacheEntry<K, V> e = new CacheEntry<K, V>(key, val,
                 stats.accessCounter.incrementAndGet());
         CacheEntry<K, V> oldCacheEntry = map.put(key, e);
         int currentSize;
         if (oldCacheEntry == null)
         {
             currentSize = stats.size.incrementAndGet();
         }
         else
         {
             currentSize = stats.size.get();
         }
         if (islive)
         {
             stats.putCounter.incrementAndGet();
         }
         else
         {
             stats.nonLivePutCounter.incrementAndGet();
         }
 
         // Check if we need to clear out old entries from the cache.
         // isCleaning variable is checked instead of markAndSweepLock.isLocked()
         // for performance because every put invokation will check until
         // the size is back to an acceptable level.
         //
         // There is a race between the check and the call to markAndSweep, but
         // it's unimportant because markAndSweep actually aquires the lock or returns if it can't.
         //
         // Thread safety note: isCleaning read is piggybacked (comes after) other volatile reads
         // in this method.
         if (currentSize > upperWaterMark && !isCleaning)
         {
             if (newThreadForCleanup)
             {
                 new Thread()
                 {
                     @Override
                     public void run()
                     {
                         markAndSweep();
                     }
                 }.start();
             }
             else if (cleanupThread != null)
             {
                 cleanupThread.wakeThread();
             }
             else
             {
                 markAndSweep();
             }
         }
         return oldCacheEntry == null ? null : oldCacheEntry.value;
     }
 
     /**
      * Removes items from the cache to bring the size down
      * to an acceptable value ('acceptableWaterMark').
      * <p/>
      * It is done in two stages. In the first stage, least recently used items are evicted.
      * If, after the first stage, the cache size is still greater than 'acceptableSize'
      * config parameter, the second stage takes over.
      * <p/>
      * The second stage is more intensive and tries to bring down the cache size
      * to the 'lowerWaterMark' config parameter.
      */
     private void markAndSweep()
     {
         // if we want to keep at least 1000 entries, then timestamps of
         // current through current-1000 are guaranteed not to be the oldest (but that does
         // not mean there are 1000 entries in that group... it's acutally anywhere between
         // 1 and 1000).
         // Also, if we want to remove 500 entries, then
         // oldestEntry through oldestEntry+500 are guaranteed to be
         // removed (however many there are there).
 
         if (!markAndSweepLock.tryLock())
         {
             return;
         }
         try
         {
             long oldestEntry = this.oldestEntry;
             isCleaning = true;
             this.oldestEntry = oldestEntry; // volatile write to make isCleaning visible
 
             long timeCurrent = stats.accessCounter.get();
             int sz = stats.size.get();
 
             int numRemoved = 0;
             int numKept = 0;
             long newestEntry = timeCurrent;
             long newNewestEntry = -1;
             long newOldestEntry = Long.MAX_VALUE;
 
             int wantToKeep = lowerWaterMark;
             int wantToRemove = sz - lowerWaterMark;
 
             @SuppressWarnings("unchecked")
             // generic array's are anoying
             CacheEntry<K, V>[] eset = new CacheEntry[sz];
             int eSize = 0;
 
             // System.out.println("newestEntry="+newestEntry + " oldestEntry="+oldestEntry);
             // System.out.println("items removed:" + numRemoved + " numKept=" + numKept + 
             //    " esetSz="+ eSize + " sz-numRemoved=" + (sz-numRemoved));
 
             for (CacheEntry<K, V> ce : map.values())
             {
                 // set lastAccessedCopy to avoid more volatile reads
                 ce.lastAccessedCopy = ce.lastAccessed;
                 long thisEntry = ce.lastAccessedCopy;
 
                 // since the wantToKeep group is likely to be bigger than wantToRemove, check it first
                 if (thisEntry > newestEntry - wantToKeep)
                 {
                     // this entry is guaranteed not to be in the bottom
                     // group, so do nothing.
                     numKept++;
                     newOldestEntry = Math.min(thisEntry, newOldestEntry);
                 }
                 else if (thisEntry < oldestEntry + wantToRemove)
                 { // entry in bottom group?
                     // this entry is guaranteed to be in the bottom group
                     // so immediately remove it from the map.
                     evictEntry(ce.key);
                     numRemoved++;
                 }
                 else
                 {
                     // This entry *could* be in the bottom group.
                     // Collect these entries to avoid another full pass... this is wasted
                     // effort if enough entries are normally removed in this first pass.
                     // An alternate impl could make a full second pass.
                     if (eSize < eset.length - 1)
                     {
                         eset[eSize++] = ce;
                         newNewestEntry = Math.max(thisEntry, newNewestEntry);
                         newOldestEntry = Math.min(thisEntry, newOldestEntry);
                     }
                 }
             }
 
             // System.out.println("items removed:" + numRemoved + " numKept=" + numKept + 
             //    " esetSz="+ eSize + " sz-numRemoved=" + (sz-numRemoved));
             // TODO: allow this to be customized in the constructor?
             int numPasses = 1; // maximum number of linear passes over the data
 
             // if we didn't remove enough entries, then make more passes
             // over the values we collected, with updated min and max values.
             while (sz - numRemoved > acceptableWaterMark && --numPasses >= 0)
             {
 
                 oldestEntry = newOldestEntry == Long.MAX_VALUE ? oldestEntry
                         : newOldestEntry;
                 newOldestEntry = Long.MAX_VALUE;
                 newestEntry = newNewestEntry;
                 newNewestEntry = -1;
                 wantToKeep = lowerWaterMark - numKept;
                 wantToRemove = sz - lowerWaterMark - numRemoved;
 
                 // iterate backward to make it easy to remove items.
                 for (int i = eSize - 1; i >= 0; i--)
                 {
                     CacheEntry<K, V> ce = eset[i];
                     long thisEntry = ce.lastAccessedCopy;
 
                     if (thisEntry > newestEntry - wantToKeep)
                     {
                         // this entry is guaranteed not to be in the bottom
                         // group, so do nothing but remove it from the eset.
                         numKept++;
                         // remove the entry by moving the last element to it's position
                         eset[i] = eset[eSize - 1];
                         eSize--;
 
                         newOldestEntry = Math.min(thisEntry, newOldestEntry);
 
                     }
                     else if (thisEntry < oldestEntry + wantToRemove)
                     { // entry in bottom group?
 
                         // this entry is guaranteed to be in the bottom group
                         // so immediately remove it from the map.
                         evictEntry(ce.key);
                         numRemoved++;
 
                         // remove the entry by moving the last element to it's position
                         eset[i] = eset[eSize - 1];
                         eSize--;
                     }
                     else
                     {
                         // This entry *could* be in the bottom group, so keep it in the eset,
                         // and update the stats.
                         newNewestEntry = Math.max(thisEntry, newNewestEntry);
                         newOldestEntry = Math.min(thisEntry, newOldestEntry);
                     }
                 }
                 // System.out.println("items removed:" + numRemoved + " numKept=" + 
                 //    numKept + " esetSz="+ eSize + " sz-numRemoved=" + (sz-numRemoved));
             }
 
             // if we still didn't remove enough entries, then make another pass while
             // inserting into a priority queue
             if (sz - numRemoved > acceptableWaterMark)
             {
 
                 oldestEntry = newOldestEntry == Long.MAX_VALUE ? oldestEntry
                         : newOldestEntry;
                 newOldestEntry = Long.MAX_VALUE;
                 newestEntry = newNewestEntry;
                 newNewestEntry = -1;
                 wantToKeep = lowerWaterMark - numKept;
                 wantToRemove = sz - lowerWaterMark - numRemoved;
 
                 PQueue<K, V> queue = new PQueue<K, V>(wantToRemove);
 
                 for (int i = eSize - 1; i >= 0; i--)
                 {
                     CacheEntry<K, V> ce = eset[i];
                     long thisEntry = ce.lastAccessedCopy;
 
                     if (thisEntry > newestEntry - wantToKeep)
                     {
                         // this entry is guaranteed not to be in the bottom
                         // group, so do nothing but remove it from the eset.
                         numKept++;
                         // removal not necessary on last pass.
                         // eset[i] = eset[eSize-1];
                         // eSize--;
 
                         newOldestEntry = Math.min(thisEntry, newOldestEntry);
 
                     }
                     else if (thisEntry < oldestEntry + wantToRemove)
                     { // entry in bottom group?
                         // this entry is guaranteed to be in the bottom group
                         // so immediately remove it.
                         evictEntry(ce.key);
                         numRemoved++;
 
                         // removal not necessary on last pass.
                         // eset[i] = eset[eSize-1];
                         // eSize--;
                     }
                     else
                     {
                         // This entry *could* be in the bottom group.
                         // add it to the priority queue
 
                         // everything in the priority queue will be removed, so keep track of
                         // the lowest value that ever comes back out of the queue.
 
                         // first reduce the size of the priority queue to account for
                         // the number of items we have already removed while executing
                         // this loop so far.
                         queue.myMaxSize = sz - lowerWaterMark - numRemoved;
                         while (queue.size() > queue.myMaxSize
                                 && queue.size() > 0)
                         {
                             CacheEntry otherEntry = (CacheEntry) queue.pop();
                             newOldestEntry = Math
                                     .min(otherEntry.lastAccessedCopy,
                                             newOldestEntry);
                         }
                         if (queue.myMaxSize <= 0)
                         {
                             break;
                         }
 
                         Object o = queue.myInsertWithOverflow(ce);
                         if (o != null)
                         {
                             newOldestEntry = Math.min(
                                     ((CacheEntry) o).lastAccessedCopy,
                                     newOldestEntry);
                         }
                     }
                 }
 
                 // Now delete everything in the priority queue.
                 // avoid using pop() since order doesn't matter anymore
                 for (CacheEntry<K, V> ce : queue.getValues())
                 {
                     if (ce == null)
                     {
                         continue;
                     }
                     evictEntry(ce.key);
                     numRemoved++;
                 }
 
                 // System.out.println("items removed:" + numRemoved + " numKept=" + numKept + 
                 //    " initialQueueSize="+ wantToRemove + " finalQueueSize=" + 
                 //      queue.size() + " sz-numRemoved=" + (sz-numRemoved));
             }
 
             oldestEntry = newOldestEntry == Long.MAX_VALUE ? oldestEntry
                     : newOldestEntry;
             this.oldestEntry = oldestEntry;
         }
         finally
         {
             isCleaning = false; // set before markAndSweep.unlock() for visibility
             markAndSweepLock.unlock();
         }
     }
 
     private static class PQueue<K, V> extends PriorityQueue<CacheEntry<K, V>>
     {
         int myMaxSize;
         final Object[] heap;
 
         PQueue(int maxSz)
         {
             super(maxSz);
             heap = getHeapArray();
             myMaxSize = maxSz;
         }
 
         @SuppressWarnings("unchecked")
         Iterable<CacheEntry<K, V>> getValues()
         {
             return (Iterable) Collections.unmodifiableCollection(Arrays
                     .asList(heap));
         }
 
         @Override
         protected boolean lessThan(CacheEntry a, CacheEntry b)
         {
             // reverse the parameter order so that the queue keeps the oldest items
             return b.lastAccessedCopy < a.lastAccessedCopy;
         }
 
         // necessary because maxSize is private in base class
         @SuppressWarnings("unchecked")
         public CacheEntry<K, V> myInsertWithOverflow(CacheEntry<K, V> element)
         {
             if (size() < myMaxSize)
             {
                 add(element);
                 return null;
             }
             else if (size() > 0
                     && !lessThan(element, (CacheEntry<K, V>) heap[1]))
             {
                 CacheEntry<K, V> ret = (CacheEntry<K, V>) heap[1];
                 heap[1] = element;
                 updateTop();
                 return ret;
             }
             else
             {
                 return element;
             }
         }
     }
 
     private void evictEntry(K key)
     {
         CacheEntry<K, V> o = map.remove(key);
         if (o == null)
         {
             return;
         }
         stats.size.decrementAndGet();
         stats.evictionCounter.incrementAndGet();
         if (evictionListener != null)
         {
             evictionListener.evictedEntry(o.key, o.value);
         }
     }
 
     /**
      * Returns 'n' number of oldest accessed entries present in this cache.
      *
      * This uses a TreeSet to collect the 'n' oldest items ordered by ascending last access time
      *  and returns a LinkedHashMap containing 'n' or less than 'n' entries.
      * @param n the number of oldest items needed
      * @return a LinkedHashMap containing 'n' or less than 'n' entries
      */
     public Map<K, V> getOldestAccessedItems(int n)
     {
         Map<K, V> result = new LinkedHashMap<K, V>();
         if (n <= 0)
         {
             return result;
         }
         TreeSet<CacheEntry<K, V>> tree = new TreeSet<CacheEntry<K, V>>();
         markAndSweepLock.lock();
         try
         {
             for (Map.Entry<Object, CacheEntry<K, V>> entry : map.entrySet())
             {
                 CacheEntry<K, V> ce = entry.getValue();
                 ce.lastAccessedCopy = ce.lastAccessed;
                 if (tree.size() < n)
                 {
                     tree.add(ce);
                 }
                 else
                 {
                     if (ce.lastAccessedCopy < tree.first().lastAccessedCopy)
                     {
                         tree.remove(tree.first());
                         tree.add(ce);
                     }
                 }
             }
         }
         finally
         {
             markAndSweepLock.unlock();
         }
         for (CacheEntry<K, V> e : tree)
         {
             result.put(e.key, e.value);
         }
         return result;
     }
 
     public Map<K, V> getLatestAccessedItems(int n)
     {
         Map<K, V> result = new LinkedHashMap<K, V>();
         if (n <= 0)
         {
             return result;
         }
         TreeSet<CacheEntry<K, V>> tree = new TreeSet<CacheEntry<K, V>>();
         // we need to grab the lock since we are changing lastAccessedCopy
         markAndSweepLock.lock();
         try
         {
             for (Map.Entry<Object, CacheEntry<K, V>> entry : map.entrySet())
             {
                 CacheEntry<K, V> ce = entry.getValue();
                 ce.lastAccessedCopy = ce.lastAccessed;
                 if (tree.size() < n)
                 {
                     tree.add(ce);
                 }
                 else
                 {
                     if (ce.lastAccessedCopy > tree.last().lastAccessedCopy)
                     {
                         tree.remove(tree.last());
                         tree.add(ce);
                     }
                 }
             }
         }
         finally
         {
             markAndSweepLock.unlock();
         }
         for (CacheEntry<K, V> e : tree)
         {
             result.put(e.key, e.value);
         }
         return result;
     }
 
     public int size()
     {
         return stats.size.get();
     }
 
     public void clear()
     {
         map.clear();
     }
 
     public Map<Object, CacheEntry<K, V>> getMap()
     {
         return map;
     }
 
     private static class CacheEntry<K, V> implements
             Comparable<CacheEntry<K, V>>
     {
         K key;
         V value;
         volatile long lastAccessed = 0;
         long lastAccessedCopy = 0;
 
         public CacheEntry(K key, V value, long lastAccessed)
         {
             this.key = key;
             this.value = value;
             this.lastAccessed = lastAccessed;
         }
 
         public void setLastAccessed(long lastAccessed)
         {
             this.lastAccessed = lastAccessed;
         }
 
         public int compareTo(CacheEntry<K, V> that)
         {
             if (this.lastAccessedCopy == that.lastAccessedCopy)
             {
                 return 0;
             }
             return this.lastAccessedCopy < that.lastAccessedCopy ? 1 : -1;
         }
 
         @Override
         public int hashCode()
         {
             return value.hashCode();
         }
 
         @Override
         public boolean equals(Object obj)
         {
             return value.equals(obj);
         }
 
         @Override
         public String toString()
         {
             return "key: " + key + " value: " + value + " lastAccessed:"
                     + lastAccessed;
         }
     }
 
     private boolean isDestroyed = false;
 
     public void destroy()
     {
         try
         {
             if (cleanupThread != null)
             {
                 cleanupThread.stopThread();
             }
         }
         finally
         {
             isDestroyed = true;
         }
     }
 
     public Stats getStats()
     {
         return stats;
     }
 
     public static class Stats
     {
         private final AtomicLong accessCounter = new AtomicLong(0);
         private final AtomicLong putCounter = new AtomicLong(0);
         private final AtomicLong nonLivePutCounter = new AtomicLong(0);
         private final AtomicLong missCounter = new AtomicLong();
         private final AtomicInteger size = new AtomicInteger();
         private AtomicLong evictionCounter = new AtomicLong();
 
         public long getCumulativeLookups()
         {
             return (accessCounter.get() - putCounter.get() - nonLivePutCounter
                     .get()) + missCounter.get();
         }
 
         public long getCumulativeHits()
         {
             return accessCounter.get() - putCounter.get()
                     - nonLivePutCounter.get();
         }
 
         public long getCumulativePuts()
         {
             return putCounter.get();
         }
 
         public long getCumulativeEvictions()
         {
             return evictionCounter.get();
         }
 
         public int getCurrentSize()
         {
             return size.get();
         }
 
         public long getCumulativeNonLivePuts()
         {
             return nonLivePutCounter.get();
         }
 
         public long getCumulativeMisses()
         {
             return missCounter.get();
         }
 
         public void add(Stats other)
         {
             accessCounter.addAndGet(other.accessCounter.get());
             putCounter.addAndGet(other.putCounter.get());
             nonLivePutCounter.addAndGet(other.nonLivePutCounter.get());
             missCounter.addAndGet(other.missCounter.get());
             evictionCounter.addAndGet(other.evictionCounter.get());
             size.set(Math.max(size.get(), other.size.get()));
         }
     }
 
     public static interface EvictionListener<K, V>
     {
         public void evictedEntry(K key, V value);
     }
 
     private static class CleanupThread extends Thread
     {
         private WeakReference<ConcurrentLRUCache> cache;
 
         private boolean stop = false;
 
         public CleanupThread(ConcurrentLRUCache c)
         {
             cache = new WeakReference<ConcurrentLRUCache>(c);
         }
 
         @Override
         public void run()
         {
             while (true)
             {
                 synchronized (this)
                 {
                     if (stop)
                     {
                         break;
                     }
                     try
                     {
                         this.wait();
                     }
                     catch (InterruptedException e)
                     {
                     }
                 }
                 if (stop)
                 {
                     break;
                 }
                 ConcurrentLRUCache c = cache.get();
                 if (c == null)
                 {
                     break;
                 }
                 c.markAndSweep();
             }
         }
 
         void wakeThread()
         {
             synchronized (this)
             {
                 this.notify();
             }
         }
 
         void stopThread()
         {
             synchronized (this)
             {
                 stop = true;
                 this.notify();
             }
         }
     }
 
     @Override
     protected void finalize() throws Throwable
     {
         try
         {
             if (!isDestroyed)
             {
                 // This log message is useless, because it is not supposed to use
                 // thread cleanup strategy for this class.
                 //log.severe("ConcurrentLRUCache was not destroyed prior to finalize()," +
                 //        " indicates a bug -- POSSIBLE RESOURCE LEAK!!!");
                 destroy();
             }
         }
         finally
         {
             super.finalize();
         }
     }
 }