/**
 * 
 */
package com.browseengine.bobo.util;

import java.lang.ref.WeakReference;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

/**
 * @author "Xiaoyang Gu<xgu@linkedin.com>"
 *
 */
public class MemoryManager<T>
{
  private static final Logger log = Logger.getLogger(MemoryManager.class.getName());
  private static final long SWEEP_INTERVAL = 60000; // 1min
  private static final ReentrantLock _sweepLock = new ReentrantLock();

  private static volatile long _nextSweepTime = System.currentTimeMillis();
  private static volatile int _nextSweepIndex = 0;


  private final ConcurrentHashMap<Integer, ConcurrentLinkedQueue<WeakReference<T>>> _sizeMap = new ConcurrentHashMap<Integer, ConcurrentLinkedQueue<WeakReference<T>>>();
  private final ConcurrentLinkedQueue<WeakReference<T>> _releaseQueue = new ConcurrentLinkedQueue<WeakReference<T>>();
  private final Initializer<T> _initializer;
  private final Thread _cleanThread;
  public MemoryManager(Initializer<T> initializer)
  {
    this._initializer = initializer;
    _cleanThread = new Thread(new Runnable(){

      public void run()
      {
        WeakReference<T> weakbuf = null;
        T buf = null;
        while(true)
        {
          synchronized(MemoryManager.this)
          {
            try
            {
              MemoryManager.this.wait(200);
            } catch (InterruptedException e)
            {
              log.error(e);
            }
          }
          while((weakbuf = _releaseQueue.poll()) != null)
          {
            if ((buf = weakbuf.get())!=null)
            {
              ConcurrentLinkedQueue<WeakReference<T>> queue = _sizeMap.get(_initializer.size(buf));
              // buf is wrapped in WeakReference. this allows GC to reclaim the buffer memory
              _initializer.init(buf);// pre-initializing the buffer in parallel so we save time when it is requested later.
              queue.offer(new WeakReference<T>(buf));
              buf = null;
            }
          }
        }
      }});
    _cleanThread.start();
  }

  /**
   * @return an initialized instance of type T.
   */
  public T get(int size)
  {
    ConcurrentLinkedQueue<WeakReference<T>> queue = _sizeMap.get(size);
    if (queue==null)
    {
      queue =  new ConcurrentLinkedQueue<WeakReference<T>>();
      _sizeMap.putIfAbsent(size, queue);
      queue = _sizeMap.get(size);
    }
    while(true)
    {
      WeakReference<T> ref = (WeakReference<T>) queue.poll();
      if(ref != null)
      {
        T buf = ref.get();
        if(buf != null)
        {
//          log.info("array hit " + size);
          return buf;
        }
      }
      else
      {
//      log.info("array miss " + size);
        return _initializer.newInstance(size);
      }
    }
  }

  /**
   * return the instance to the manager after use
   * @param buf
   */
  public void release(T buf)
  {
    if(buf != null)
    {
      _releaseQueue.offer(new WeakReference<T>(buf));
      synchronized(MemoryManager.this)
      {
        MemoryManager.this.notifyAll();
      }
    }
  }

  public static interface Initializer<E>
  {
    public E newInstance(int size);
    public int size(E buf);
    public void init(E buf);
  }
}

@@ -208,27 +208,28 @@
 ent.
-locks.ReentrantLock
+atomic.AtomicInteger
 ;%0A%0Ai
@@ -435,268 +435,8 @@
 ));%0A
-  private static final long SWEEP_INTERVAL = 60000; // 1min%0A  private static final ReentrantLock _sweepLock = new ReentrantLock();%0A%0A  private static volatile long _nextSweepTime = System.currentTimeMillis();%0A  private static volatile int _nextSweepIndex = 0;%0A%0A
 %0A  p
@@ -640,32 +640,17 @@
 edQueue%3C
-WeakReference%3CT%3E
+T
 %3E _relea
@@ -685,34 +685,91 @@
 edQueue%3C
-WeakReference%3CT%3E%3E(
+T%3E();%0A  private final AtomicInteger _releaseQueueSize = new AtomicInteger(0
 );%0A  pri
@@ -1020,49 +1020,8 @@
   %7B%0A
-        WeakReference%3CT%3E weakbuf = null;%0A
     
@@ -1327,20 +1327,16 @@
  while((
-weak
 buf = _r
@@ -1380,69 +1380,8 @@
   %7B%0A
-            if ((buf = weakbuf.get())!=null)%0A            %7B%0A  
     
@@ -1486,18 +1486,16 @@
         
-  
 // buf i
@@ -1578,18 +1578,16 @@
         
- 
- 
 _initial
@@ -1696,26 +1696,24 @@
             
-  
 queue.offer(
@@ -1756,24 +1756,45 @@
     
- 
- buf = null;%0A  
+_releaseQueueSize.decrementAndGet();%0A
     
@@ -1803,33 +1803,43 @@
     %7D%0A          
-%7D
+buf = null;
 %0A        %7D%0A     
@@ -2385,51 +2385,8 @@
   %7B%0A
-//          log.info(%22array hit %22 + size);%0A
     
@@ -2444,48 +2444,8 @@
   %7B%0A
-//      log.info(%22array miss %22 + size);%0A
     
@@ -2627,79 +2627,202 @@
   if
-(buf != null)%0A    %7B%0A      _releaseQueue.offer(new WeakReference%3CT%3E(buf)
+ (_releaseQueueSize.get()%3E1000)%0A    %7B%0A      log.info(%22release queue full%22);%0A      return;%0A    %7D%0A    if(buf != null)%0A    %7B%0A      _releaseQueue.offer(buf);%0A      _releaseQueueSize.incrementAndGet(
 );%0A 

clean up memory poll