package org.apache.solr.util;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.TreeSet;
import java.util.Map;
import java.util.HashMap;

/**
 * @author Noble Paul (Noble.Paul@corp.aol.com)
 *         Date: Jul 29, 2008
 *         Time: 12:53:53 PM
 */
public class ConcurrentLRUCache {
  private AtomicLong accessCounter = new AtomicLong(0);
  private Map<Object, Entry> map;
  private int upperWaterMark,lowerWaterMark;
  private ScheduledExecutorService executorService ;

  public ConcurrentLRUCache(int size, int evictionBatchSize, boolean runCleanupThread, int delay) {
    if(size < 1) throw new IllegalArgumentException("upperWaterMark must be > 0");
    if(evictionBatchSize > size) throw new IllegalArgumentException("evictionBatchSize must be  < upperWaterMark");
//    map = new ConcurrentHashMap<Object, Entry>(size);
    map = new HashMap<Object, Entry>(size);
    this.upperWaterMark = size;
    this.lowerWaterMark =  size - evictionBatchSize;
    if(runCleanupThread){
      executorService = Executors.newSingleThreadScheduledExecutor();
      executorService.scheduleAtFixedRate(new Runnable() {
        public void run() {
          if(map.size() > lowerWaterMark) markAndSweep();
        }
      }, delay, delay, TimeUnit.SECONDS);
    }
  }
  public Object get(Object key) {
    Entry e = map.get(key);
    e.setLastAccessed(accessCounter.incrementAndGet());
    return e.value;
  }

  public void remove(Object key){
    map.remove(key);
  }

  public void put(Object key , Object val){
    Entry e = new Entry(key, val, accessCounter.incrementAndGet());
    map.put(key, e);
    if(map.size() > upperWaterMark){
      markAndSweep();
    }

  }

  private synchronized void markAndSweep() {
    System.out.println("ConcurrentLRUCache.markAndSweep");
    int itemsToBeremoved = map.size() - lowerWaterMark;
    if(itemsToBeremoved < 1) return;
    TreeSet<SortEntry> tree =  new TreeSet<SortEntry>();
    for (Map.Entry<Object, Entry> entry : map.entrySet()) {
      if(tree.size() < itemsToBeremoved) {
        tree.add(new SortEntry(entry.getValue()));
      } else  {
        long lastAccessed = entry.getValue().lastAccessed;
        if(lastAccessed < tree.first().lastAccessed){
          tree.remove(tree.first());
          tree.add(new SortEntry(entry.getValue()));
        }
      }
    }
    for (SortEntry sortEntry : tree) map.remove(sortEntry.entry.key);
  }

  private static class SortEntry implements Comparable<SortEntry>{
    Entry entry;
    final long lastAccessed;

    public SortEntry(Entry e) {
      this.entry = e;
      lastAccessed = e.lastAccessed;
    }
    public int compareTo(SortEntry that) {
      if(this.lastAccessed == that.lastAccessed) return 0;
      return this.lastAccessed < that.lastAccessed ? 1: -1;
    }

    public boolean equals(Object obj) {
      if (obj instanceof SortEntry) {
        SortEntry sortEntry = (SortEntry) obj;
        return entry.equals(sortEntry.entry);
      }
      return false;
    }
  }

  private static class Entry {
    Object key, value;
    volatile long lastAccessed = -1;


    public Entry(Object key, Object value, long lastAccessed) {
      this.key = key;
      this.value = value;
      this.lastAccessed = lastAccessed;
    }

    public synchronized void setLastAccessed(long lastAccessed) {
      this.lastAccessed = lastAccessed;
    }

    public int hashCode() {
      return key.hashCode();
    }
  }
  public void destroy(){
    if(executorService != null){
      executorService.shutdownNow();
    }
    map.clear();
    executorService = null;

  }

  public static void main(String[] args) {
    long uwm =      1000000;

    ConcurrentLRUCache cache1 = new ConcurrentLRUCache((int)uwm,10000,false,0);
    int i =0;
    for(;i<uwm;i++){
      cache1.put(i,""+i);
    }
     i=0;

    for(;i<uwm;i++){
      cache1.get(i);
    }
    cache1.destroy();

    ConcurrentLRUCache cache = new ConcurrentLRUCache((int)uwm,10000,false,0);
    long start = System.currentTimeMillis();
    i =0;
    for(;i<uwm;i++){
      cache.put(i,""+i);
    }
    System.out.println("time for puts  = " + (System.currentTimeMillis() -start));
    start = System.currentTimeMillis();
    i=0;
    for(;i<uwm;i++){
      cache.get(i);
    }
    System.out.println("time for gets = " + (System.currentTimeMillis() -start));
    i++;
    start = System.currentTimeMillis();
    cache.put(i, ""+i);
//    end = System.currentTimeMillis() -end;
    System.out.println("time for cleanup = " + (System.currentTimeMillis() -start));

  }

}