package org.apache.jcs.auxiliary.disk.indexed;
   
   /*
    * 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.
   */
  
  import java.io.File;
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.io.Serializable;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Comparator;
  import java.util.ConcurrentModificationException;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.jcs.auxiliary.AuxiliaryCacheAttributes;
  import org.apache.jcs.auxiliary.disk.AbstractDiskCache;
  import org.apache.jcs.auxiliary.disk.LRUMapJCS;
  import org.apache.jcs.engine.CacheConstants;
  import org.apache.jcs.engine.behavior.ICacheElement;
  import org.apache.jcs.engine.behavior.IElementSerializer;
  import org.apache.jcs.engine.control.group.GroupAttrName;
  import org.apache.jcs.engine.control.group.GroupId;
  import org.apache.jcs.engine.logging.behavior.ICacheEvent;
  import org.apache.jcs.engine.logging.behavior.ICacheEventLogger;
  import org.apache.jcs.engine.stats.StatElement;
  import org.apache.jcs.engine.stats.Stats;
  import org.apache.jcs.engine.stats.behavior.IStatElement;
  import org.apache.jcs.engine.stats.behavior.IStats;
  import org.apache.jcs.utils.struct.SortedPreferentialArray;
  import org.apache.jcs.utils.timing.ElapsedTimer;
  
  /**
   * Disk cache that uses a RandomAccessFile with keys stored in memory. The maximum number of keys
   * stored in memory is configurable. The disk cache tries to recycle spots on disk to limit file
   * expansion.
   */
  public class IndexedDiskCache<K extends Serializable, V extends Serializable>
      extends AbstractDiskCache<K, V>
  {
      /** Don't change */
      private static final long serialVersionUID = -265035607729729629L;
  
      /** The logger */
      protected static final Log log = LogFactory.getLog( IndexedDiskCache.class );
  
      /** Cache name used in log messages */
      protected final String logCacheName;
  
      /** The name of the file where the data is stored */
      private final String fileName;
  
      /** The IndexedDisk manages reads and writes to the data file. */
      private IndexedDisk dataFile;
  
      /** The IndexedDisk manages reads and writes to the key file. */
      private IndexedDisk keyFile;
  
      /** Map containing the keys and disk offsets. */
      private Map<K, IndexedDiskElementDescriptor> keyHash;
  
      /** The maximum number of keys that we will keep in memory. */
      private final int maxKeySize;
  
      /** A handle on the data file. */
      private File rafDir;
  
      /** Should we keep adding to the recycle bin. False during optimization. */
      boolean doRecycle = true;
  
      /** Should we optimize real time */
      boolean isRealTimeOptimizationEnabled = true;
  
     /** Should we optimize on shutdown. */
     boolean isShutdownOptimizationEnabled = true;
 
     /** are we currently optimizing the files */
     boolean isOptimizing = false;
 
     /** The number of times the file has been optimized. */
     private int timesOptimized = 0;
 
     /** The thread optimizing the file. */
     protected volatile Thread currentOptimizationThread;
 
     /** used for counting the number of requests */
     private int removeCount = 0;
 
     /** Should we queue puts. True when optimizing. We write the queue post optimization. */
     private boolean queueInput = false;
 
     /** list where puts made during optimization are made */
     private final LinkedList<IndexedDiskElementDescriptor> queuedPutList =
         new LinkedList<IndexedDiskElementDescriptor>();
 
     /** RECYLCE BIN -- array of empty spots */
     private SortedPreferentialArray<IndexedDiskElementDescriptor> recycle;
 
     /** User configurable parameters */
     private final IndexedDiskCacheAttributes cattr;
 
     /** How many slots have we recycled. */
     private int recycleCnt = 0;
 
     /** How many items were there on startup. */
     private int startupSize = 0;
 
     /** the number of bytes free on disk. */
     private long bytesFree = 0;
 
     /** simple stat */
     private AtomicInteger hitCount = new AtomicInteger(0);
 
     /**
      * Use this lock to synchronize reads and writes to the underlying storage mechanism.
      */
     protected ReentrantReadWriteLock storageLock = new ReentrantReadWriteLock();
 
     /**
      * Constructor for the DiskCache object.
      * <p>
      * @param cacheAttributes
      */
     public IndexedDiskCache( IndexedDiskCacheAttributes cacheAttributes )
     {
         this( cacheAttributes, null );
     }
 
     /**
      * Constructor for the DiskCache object.
      * <p>
      * @param cattr
      * @param elementSerializer used if supplied, the super's super will not set a null
      */
     public IndexedDiskCache( IndexedDiskCacheAttributes cattr, IElementSerializer elementSerializer )
     {
         super( cattr );
 
         setElementSerializer( elementSerializer );
 
         this.cattr = cattr;
         this.maxKeySize = cattr.getMaxKeySize();
         this.isRealTimeOptimizationEnabled = cattr.getOptimizeAtRemoveCount() > 0;
         this.isShutdownOptimizationEnabled = cattr.isOptimizeOnShutdown();
         this.logCacheName = "Region [" + getCacheName() + "] ";
         this.fileName = getCacheName();
 
         try
         {
             initializeFileSystem( cattr );
 
             initializeKeysAndData( cattr );
 
             initializeRecycleBin();
 
             // Initialization finished successfully, so set alive to true.
             alive = true;
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "Indexed Disk Cache is alive." );
             }
 
             // TODO: Should we improve detection of whether or not the file should be optimized.
             if ( isRealTimeOptimizationEnabled && keyHash.size() > 0 )
             {
                 // Kick off a real time optimization, in case we didn't do a final optimization.
                 doOptimizeRealTime();
             }
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Failure initializing for fileName: " + fileName + " and directory: "
                 + this.rafDir.getAbsolutePath(), e );
         }
     }
 
     /**
      * Tries to create the root directory if it does not already exist.
      * <p>
      * @param cattr
      */
     private void initializeFileSystem( IndexedDiskCacheAttributes cattr )
     {
         String rootDirName = cattr.getDiskPath();
         this.rafDir = new File( rootDirName );
         boolean createdDirectories = this.rafDir.mkdirs();
         if ( log.isInfoEnabled() )
         {
             log.info( logCacheName + "Cache file root directory: " + rootDirName );
             log.info( logCacheName + "Created root directory: " + createdDirectories );
         }
     }
 
     /**
      * Creates the key and data disk caches.
      * <p>
      * Loads any keys if they are present and ClearDiskOnStartup is false.
      * <p>
      * @param cattr
      * @throws FileNotFoundException
      * @throws IOException
      * @throws InterruptedException
      */
     private void initializeKeysAndData( IndexedDiskCacheAttributes cattr )
         throws FileNotFoundException, IOException, InterruptedException
     {
         this.dataFile = new IndexedDisk( new File( rafDir, fileName + ".data" ), getElementSerializer() );
 
         this.keyFile = new IndexedDisk( new File( rafDir, fileName + ".key" ), getElementSerializer() );
 
         if ( cattr.isClearDiskOnStartup() )
         {
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "ClearDiskOnStartup is set to true.  Ingnoring any persisted data." );
             }
             initializeEmptyStore();
         }
         else if ( keyFile.length() > 0 )
         {
             // If the key file has contents, try to initialize the keys
             // from it. In no keys are loaded reset the data file.
             initializeStoreFromPersistedData();
         }
         else
         {
             // Otherwise start with a new empty map for the keys, and reset
             // the data file if it has contents.
             initializeEmptyStore();
         }
     }
 
     /**
      * Initializes an empty disk cache.
      * <p>
      * @throws IOException
      */
     private void initializeEmptyStore()
         throws IOException
     {
         initializeKeyMap();
 
         if ( dataFile.length() > 0 )
         {
             dataFile.reset();
         }
     }
 
     /**
      * Loads any persisted data and checks for consistency. If there is a consistency issue, the
      * files are cleared.
      * <p>
      * @throws InterruptedException
      * @throws IOException
      */
     private void initializeStoreFromPersistedData()
         throws InterruptedException, IOException
     {
         loadKeys();
 
         if ( keyHash.size() == 0 )
         {
             dataFile.reset();
         }
         else
         {
             boolean isOk = checkKeyDataConsistency( false );
             if ( !isOk )
             {
                 keyHash.clear();
                 keyFile.reset();
                 dataFile.reset();
                 log.warn( logCacheName + "Corruption detected.  Reseting data and keys files." );
             }
             else
             {
                 synchronized (this)
                 {
                     startupSize = keyHash.size();
                 }
             }
         }
     }
 
     /**
      * Loads the keys from the .key file. The keys are stored in a HashMap on disk. This is
      * converted into a LRUMap.
      * <p>
      * @throws InterruptedException
      */
     protected void loadKeys()
         throws InterruptedException
     {
         if ( log.isDebugEnabled() )
         {
             log.debug( logCacheName + "Loading keys for " + keyFile.toString() );
         }
 
         storageLock.writeLock().lock();
 
         try
         {
             // create a key map to use.
             initializeKeyMap();
 
             HashMap<K, IndexedDiskElementDescriptor> keys =
                 keyFile.readObject( new IndexedDiskElementDescriptor( 0, (int) keyFile.length()
                 - IndexedDisk.HEADER_SIZE_BYTES ) );
 
             if ( keys != null )
             {
                 if ( log.isDebugEnabled() )
                 {
                     log.debug( logCacheName + "Found " + keys.size() + " in keys file." );
                 }
 
                 keyHash.putAll( keys );
 
                 if ( log.isInfoEnabled() )
                 {
                     log.info( logCacheName + "Loaded keys from [" + fileName + "], key count: " + keyHash.size()
                         + "; up to " + maxKeySize + " will be available." );
                 }
             }
 
             if ( log.isDebugEnabled() )
             {
                 dump( false );
             }
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Problem loading keys for file " + fileName, e );
         }
         finally
         {
             storageLock.writeLock().unlock();
         }
     }
 
     /**
      * Check for minimal consistency between the keys and the datafile. Makes sure no starting
      * positions in the keys exceed the file length.
      * <p>
      * The caller should take the appropriate action if the keys and data are not consistent.
      * @param checkForDedOverlaps if <code>true</code>, do a more thorough check by checking for
      *            data overlap
      * @return <code>true</code> if the test passes
      */
     private boolean checkKeyDataConsistency( boolean checkForDedOverlaps )
     {
         ElapsedTimer timer = new ElapsedTimer();
         log.debug( logCacheName + "Performing inital consistency check" );
 
         boolean isOk = true;
         long fileLength = 0;
         try
         {
             fileLength = dataFile.length();
 
             for (Map.Entry<K, IndexedDiskElementDescriptor> e : keyHash.entrySet())
             {
                 IndexedDiskElementDescriptor ded = e.getValue();
 
                 isOk = ( ded.pos + IndexedDisk.HEADER_SIZE_BYTES + ded.len <= fileLength );
 
                 if ( !isOk )
                 {
                     log.warn( logCacheName + "The dataFile is corrupted!" + "\n raf.length() = " + fileLength
                         + "\n ded.pos = " + ded.pos );
                     break;
                 }
             }
 
             if ( isOk && checkForDedOverlaps )
             {
                 isOk = checkForDedOverlaps( createPositionSortedDescriptorList() );
             }
         }
         catch ( Exception e )
         {
             log.error( e );
             isOk = false;
         }
 
         if ( log.isInfoEnabled() )
         {
             log.info( logCacheName + "Finished inital consistency check, isOk = " + isOk + " in "
                 + timer.getElapsedTimeString() );
         }
 
         return isOk;
     }
 
     /**
      * Detects any overlapping elements. This expects a sorted list.
      * <p>
      * The total length of an item is IndexedDisk.RECORD_HEADER + ded.len.
      * <p>
      * @param sortedDescriptors
      * @return false if there are overlaps.
      */
     protected boolean checkForDedOverlaps( IndexedDiskElementDescriptor[] sortedDescriptors )
     {
         long start = System.currentTimeMillis();
         boolean isOk = true;
         long expectedNextPos = 0;
         for ( int i = 0; i < sortedDescriptors.length; i++ )
         {
             IndexedDiskElementDescriptor ded = sortedDescriptors[i];
             if ( expectedNextPos > ded.pos )
             {
                 log.error( logCacheName + "Corrupt file: overlapping deds " + ded );
                 isOk = false;
                 break;
             }
             else
             {
                 expectedNextPos = ded.pos + IndexedDisk.HEADER_SIZE_BYTES + ded.len;
             }
         }
         long end = System.currentTimeMillis();
         if ( log.isDebugEnabled() )
         {
             log.debug( logCacheName + "Check for DED overlaps took " + ( end - start ) + " ms." );
         }
 
         return isOk;
     }
 
     /**
      * Saves key file to disk. This converts the LRUMap to a HashMap for deserialization.
      */
     protected void saveKeys()
     {
         try
         {
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "Saving keys to: " + fileName + ", key count: " + keyHash.size() );
             }
 
             keyFile.reset();
 
             HashMap<K, IndexedDiskElementDescriptor> keys =
                 new HashMap<K, IndexedDiskElementDescriptor>();
             keys.putAll( keyHash );
 
             if ( keys.size() > 0 )
             {
                 keyFile.writeObject( keys, 0 );
             }
 
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "Finished saving keys." );
             }
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Problem storing keys.", e );
         }
     }
 
     /**
      * Update the disk cache. Called from the Queue. Makes sure the Item has not been retrieved from
      * purgatory while in queue for disk. Remove items from purgatory when they go to disk.
      * <p>
      * @param ce The ICacheElement<K, V> to put to disk.
      */
     @Override
     protected void processUpdate( ICacheElement<K, V> ce )
     {
         if ( !alive )
         {
             log.error( logCacheName + "No longer alive; aborting put of key = " + ce.getKey() );
             return;
         }
 
         if ( log.isDebugEnabled() )
         {
             log.debug( logCacheName + "Storing element on disk, key: " + ce.getKey() );
         }
 
         IndexedDiskElementDescriptor ded = null;
 
         // old element with same key
         IndexedDiskElementDescriptor old = null;
 
         try
         {
             byte[] data = getElementSerializer().serialize( ce );
 
             // make sure this only locks for one particular cache region
             storageLock.writeLock().lock();
             try
             {
                 old = keyHash.get( ce.getKey() );
 
                 // Item with the same key already exists in file.
                 // Try to reuse the location if possible.
                 if ( old != null && data.length <= old.len )
                 {
                     // Reuse the old ded. The defrag relies on ded updates by reference, not
                     // replacement.
                     ded = old;
                     ded.len = data.length;
                 }
                 else
                 {
                     // we need this to compare in the recycle bin
                     ded = new IndexedDiskElementDescriptor( dataFile.length(), data.length );
 
                     if ( doRecycle )
                     {
                         IndexedDiskElementDescriptor rep = recycle
                             .takeNearestLargerOrEqual( ded );
                         if ( rep != null )
                         {
                             ded = rep;
                             ded.len = data.length;
                             recycleCnt++;
                             this.adjustBytesFree( ded, false );
                             if ( log.isDebugEnabled() )
                             {
                                 log.debug( logCacheName + "using recycled ded " + ded.pos + " rep.len = " + rep.len
                                     + " ded.len = " + ded.len );
                             }
                         }
                     }
 
                     // Put it in the map
                     keyHash.put( ce.getKey(), ded );
 
                     if ( queueInput )
                     {
                         queuedPutList.add( ded );
                         if ( log.isDebugEnabled() )
                         {
                             log.debug( logCacheName + "added to queued put list." + queuedPutList.size() );
                         }
                     }
 
                     // add the old slot to the recycle bin
                     if ( old != null )
                     {
                         addToRecycleBin( old );
                     }
                 }
 
                 dataFile.write( ded, data );
             }
             finally
             {
                 storageLock.writeLock().unlock();
             }
 
             if ( log.isDebugEnabled() )
             {
                 log.debug( logCacheName + "Put to file: " + fileName + ", key: " + ce.getKey() + ", position: "
                     + ded.pos + ", size: " + ded.len );
             }
         }
         catch ( ConcurrentModificationException cme )
         {
             // do nothing, this means it has gone back to memory mid
             // serialization
             if ( log.isDebugEnabled() )
             {
                 // this shouldn't be possible
                 log.debug( logCacheName + "Caught ConcurrentModificationException." + cme );
             }
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Failure updating element, key: " + ce.getKey() + " old: " + old, e );
         }
     }
 
     /**
      * Gets the key, then goes to disk to get the object.
      * <p>
      * @param key
      * @return ICacheElement<K, V> or null
      * @see AbstractDiskCache#doGet
      */
     @Override
     protected ICacheElement<K, V> processGet( K key )
     {
         if ( !alive )
         {
             log.error( logCacheName + "No longer alive so returning null for key = " + key );
             return null;
         }
 
         if ( log.isDebugEnabled() )
         {
             log.debug( logCacheName + "Trying to get from disk: " + key );
         }
 
         ICacheElement<K, V> object = null;
         try
         {
             storageLock.readLock().lock();
             try
             {
                 object = readElement( key );
             }
             finally
             {
                 storageLock.readLock().unlock();
             }
 
             if ( object != null )
             {
                 hitCount.incrementAndGet();
             }
         }
         catch ( IOException ioe )
         {
             log.error( logCacheName + "Failure getting from disk, key = " + key, ioe );
             reset();
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Failure getting from disk, key = " + key, e );
         }
         return object;
     }
 
     /**
      * Gets matching items from the cache.
      * <p>
      * @param pattern
      * @return a map of K key to ICacheElement<K, V> element, or an empty map if there is no
      *         data in cache matching keys
      */
     @Override
     public Map<K, ICacheElement<K, V>> processGetMatching( String pattern )
     {
         Map<K, ICacheElement<K, V>> elements = new HashMap<K, ICacheElement<K, V>>();
         try
         {
             Set<K> keyArray = null;
             storageLock.readLock().lock();
             try
             {
                 keyArray = new HashSet<K>(keyHash.keySet());
             }
             finally
             {
                 storageLock.readLock().unlock();
             }
 
             Set<K> matchingKeys = getKeyMatcher().getMatchingKeysFromArray( pattern, keyArray );
 
             for (K key : matchingKeys)
             {
                 ICacheElement<K, V> element = processGet( key );
                 if ( element != null )
                 {
                     elements.put( key, element );
                 }
             }
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Failure getting matching from disk, pattern = " + pattern, e );
         }
         return elements;
     }
 
     /**
      * Reads the item from disk.
      * <p>
      * @param key
      * @return ICacheElement
      * @throws IOException
      */
     private ICacheElement<K, V> readElement( K key )
         throws IOException
     {
         ICacheElement<K, V> object = null;
 
         IndexedDiskElementDescriptor ded = keyHash.get( key );
 
         if ( ded != null )
         {
             if ( log.isDebugEnabled() )
             {
                 log.debug( logCacheName + "Found on disk, key: " + key );
             }
             try
             {
                 ICacheElement<K, V> readObject = dataFile.readObject( ded );
                 object = readObject;
                 // TODO consider checking key equality and throwing if there is a failure
             }
             catch ( IOException e )
             {
                 log.error( logCacheName + "IO Exception, Problem reading object from file", e );
                 throw e;
             }
             catch ( Exception e )
             {
                 log.error( logCacheName + "Exception, Problem reading object from file", e );
                 throw new IOException( logCacheName + "Problem reading object from disk. " + e.getMessage() );
             }
         }
 
         return object;
     }
 
     /**
      * Gets the group keys from the disk.
      * <p>
      * @see org.apache.jcs.auxiliary.AuxiliaryCache#getGroupKeys(java.lang.String)
      */
     @Override
     public Set<K> getGroupKeys( String groupName )
     {
         GroupId groupId = new GroupId( cacheName, groupName );
         HashSet<K> keys = new HashSet<K>();
         try
         {
             storageLock.readLock().lock();
 
             for (Serializable k : keyHash.keySet())
             {
                 if ( k instanceof GroupAttrName && ( (GroupAttrName<?>) k ).groupId.equals( groupId ) )
                 {
                     @SuppressWarnings("unchecked") // Type checked with instanceof
                     GroupAttrName<K> groupAttrName = (GroupAttrName<K>) k;
                     keys.add( groupAttrName.attrName );
                 }
             }
         }
         finally
         {
             storageLock.readLock().unlock();
         }
 
         return keys;
     }
 
     /**
      * Gets the group names from the disk.
      * <p>
      * @see org.apache.jcs.auxiliary.AuxiliaryCache#getGroupKeys(java.lang.String)
      */
     @Override
     public Set<String> getGroupNames()
     {
         HashSet<String> names = new HashSet<String>();
         try
         {
             storageLock.readLock().lock();
 
             for (K k : keyHash.keySet())
             {
                 if ( k instanceof GroupAttrName )
                 {
                     @SuppressWarnings("unchecked") // Type checked with instanceof
                     GroupAttrName<K> groupAttrName = (GroupAttrName<K>) k;
                     names.add( groupAttrName.groupId.groupName );
                 }
             }
         }
         finally
         {
             storageLock.readLock().unlock();
         }
 
         return names;
     }
 
     /**
      * Returns true if the removal was successful; or false if there is nothing to remove. Current
      * implementation always result in a disk orphan.
      * <p>
      * @return true if at least one item was removed.
      * @param key
      */
     @Override
     protected boolean processRemove( K key )
     {
         if ( !alive )
         {
             log.error( logCacheName + "No longer alive so returning false for key = " + key );
             return false;
         }
 
         if ( key == null )
         {
             return false;
         }
 
         boolean reset = false;
         boolean removed = false;
         try
         {
             storageLock.writeLock().lock();
 
             if ( key instanceof String && key.toString().endsWith( CacheConstants.NAME_COMPONENT_DELIMITER ) )
             {
                 removed = performPartialKeyRemoval( (String) key );
             }
             else if ( key instanceof GroupAttrName && ((GroupAttrName<?>)key).attrName == null )
             {
                 removed = performGroupRemoval( ((GroupAttrName<?>)key).groupId );
             }
             else
             {
                 removed = performSingleKeyRemoval( key );
             }
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Problem removing element.", e );
             reset = true;
         }
         finally
         {
             storageLock.writeLock().unlock();
         }
 
         if ( reset )
         {
             reset();
         }
 
         // this increments the remove count.
         // there is no reason to call this if an item was not removed.
         if ( removed )
         {
             doOptimizeRealTime();
         }
 
         return removed;
     }
 
     /**
      * Iterates over the keyset. Builds a list of matches. Removes all the keys in the list. Does
      * not remove via the iterator, since the map impl may not support it.
      * <p>
      * This operates under a lock obtained in doRemove().
      * <p>
      * @param key
      * @return true if there was a match
      */
     private boolean performPartialKeyRemoval( String key )
     {
         boolean removed = false;
 
         // remove all keys of the same name hierarchy.
         List<K> itemsToRemove = new LinkedList<K>();
 
         for (K k : keyHash.keySet())
         {
             if ( k instanceof String && k.toString().startsWith( key.toString() ) )
             {
                 itemsToRemove.add( k );
             }
         }
 
         // remove matches.
         for (K fullKey : itemsToRemove)
         {
             // Don't add to recycle bin here
             // https://issues.apache.org/jira/browse/JCS-67
             performSingleKeyRemoval( fullKey );
             removed = true;
             // TODO this needs to update the remove count separately
         }
 
         return removed;
     }
 
     /**
      * Remove all elements from the group. This does not use the iterator to remove. It builds a
      * list of group elements and then removes them one by one.
      * <p>
      * This operates under a lock obtained in doRemove().
      * <p>
      * @param key
      * @return true if an element was removed
      */
     private boolean performGroupRemoval( GroupId key )
     {
         boolean removed = false;
 
         // remove all keys of the same name group.
         List<K> itemsToRemove = new LinkedList<K>();
 
         // remove all keys of the same name hierarchy.
         for (K k : keyHash.keySet())
         {
             if ( k instanceof GroupAttrName && ( (GroupAttrName<?>) k ).groupId.equals( key ) )
             {
                 itemsToRemove.add( k );
             }
         }
 
         // remove matches.
         for (K fullKey : itemsToRemove)
         {
             // Don't add to recycle bin here
             // https://issues.apache.org/jira/browse/JCS-67
             performSingleKeyRemoval( fullKey );
             removed = true;
             // TODO this needs to update the remove count separately
         }
 
         return removed;
     }
 
     /**
      * Removes an individual key from the cache.
      * <p>
      * This operates under a lock obtained in doRemove().
      * <p>
      * @param key
      * @return true if an item was removed.
      */
     private boolean performSingleKeyRemoval( K key )
     {
         boolean removed;
         // remove single item.
         IndexedDiskElementDescriptor ded = keyHash.remove( key );
         removed = ( ded != null );
         addToRecycleBin( ded );
 
         if ( log.isDebugEnabled() )
         {
             log.debug( logCacheName + "Disk removal: Removed from key hash, key [" + key + "] removed = " + removed );
         }
         return removed;
     }
 
     /**
      * Remove all the items from the disk cache by reseting everything.
      */
     @Override
     public void processRemoveAll()
     {
         ICacheEvent<String> cacheEvent = createICacheEvent( cacheName, "all", ICacheEventLogger.REMOVEALL_EVENT );
         try
         {
             reset();
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Problem removing all.", e );
             reset();
         }
         finally
         {
             logICacheEvent( cacheEvent );
         }
     }
 
     /**
      * Reset effectively clears the disk cache, creating new files, recyclebins, and keymaps.
      * <p>
      * It can be used to handle errors by last resort, force content update, or removeall.
      */
     private void reset()
     {
         if ( log.isWarnEnabled() )
         {
             log.warn( logCacheName + "Reseting cache" );
         }
 
         try
         {
             storageLock.writeLock().lock();
 
             if ( dataFile != null )
             {
                 dataFile.close();
             }
             File dataFileTemp = new File( rafDir, fileName + ".data" );
             dataFileTemp.delete();
 
             if ( keyFile != null )
             {
                 keyFile.close();
             }
             File keyFileTemp = new File( rafDir, fileName + ".key" );
             keyFileTemp.delete();
 
             dataFile = new IndexedDisk( new File( rafDir, fileName + ".data" ), getElementSerializer() );
 
             keyFile = new IndexedDisk( new File( rafDir, fileName + ".key" ), getElementSerializer() );
 
             initializeRecycleBin();
 
             initializeKeyMap();
         }
         catch ( Exception e )
         {
             log.error( logCacheName + "Failure reseting state", e );
         }
         finally
         {
             storageLock.writeLock().unlock();
         }
     }
 
     /**
      * If the maxKeySize is < 0, use 5000, no way to have an unlimited recycle bin right now, or one
      * less than the mazKeySize.
      */
     private void initializeRecycleBin()
     {
         int recycleBinSize = cattr.getMaxRecycleBinSize() >= 0 ? cattr.getMaxRecycleBinSize() : 0;
         recycle = new SortedPreferentialArray<IndexedDiskElementDescriptor>( recycleBinSize );
         if ( log.isDebugEnabled() )
         {
             log.debug( logCacheName + "Set recycle max Size to MaxRecycleBinSize: '" + recycleBinSize + "'" );
         }
     }
 
     /**
      * Create the map for keys that contain the index position on disk.
      */
     private void initializeKeyMap()
     {
         keyHash = null;
         if ( maxKeySize >= 0 )
         {
             keyHash = new LRUMap( maxKeySize );
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "Set maxKeySize to: '" + maxKeySize + "'" );
             }
         }
         else
         {
             // If no max size, use a plain map for memory and processing efficiency.
             keyHash = new HashMap<K, IndexedDiskElementDescriptor>();
             // keyHash = Collections.synchronizedMap( new HashMap() );
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "Set maxKeySize to unlimited'" );
             }
         }
     }
 
     /**
      * Dispose of the disk cache in a background thread. Joins against this thread to put a cap on
      * the disposal time.
      * <p>
      * TODO make dispose window configurable.
      */
     @Override
     public void processDispose()
     {
         ICacheEvent<String> cacheEvent = createICacheEvent( cacheName, "none", ICacheEventLogger.DISPOSE_EVENT );
         try
         {
             Runnable disR = new Runnable()
             {
                 public void run()
                 {
                     disposeInternal();
                 }
             };
             Thread t = new Thread( disR, "IndexedDiskCache-DisposalThread" );
             t.start();
             // wait up to 60 seconds for dispose and then quit if not done.
             try
             {
                 t.join( 60 * 1000 );
             }
             catch ( InterruptedException ex )
             {
                 log.error( logCacheName + "Interrupted while waiting for disposal thread to finish.", ex );
             }
         }
         finally
         {
             logICacheEvent( cacheEvent );
         }
     }
 
     /**
      * Internal method that handles the disposal.
      */
     protected void disposeInternal()
     {
         if ( !alive )
         {
             log.error( logCacheName + "Not alive and dispose was called, filename: " + fileName );
             return;
         }
 
         // Prevents any interaction with the cache while we're shutting down.
         alive = false;
 
         Thread optimizationThread = currentOptimizationThread;
         if ( isRealTimeOptimizationEnabled && optimizationThread != null )
         {
             // Join with the current optimization thread.
             if ( log.isDebugEnabled() )
             {
                 log.debug( logCacheName + "In dispose, optimization already " + "in progress; waiting for completion." );
             }
             try
             {
                 optimizationThread.join();
             }
             catch ( InterruptedException e )
             {
                 log.error( logCacheName + "Unable to join current optimization thread.", e );
             }
         }
         else if ( isShutdownOptimizationEnabled && this.getBytesFree() > 0 )
         {
             optimizeFile();
         }
 
         saveKeys();
 
         try
         {
             if ( log.isDebugEnabled() )
             {
                 log.debug( logCacheName + "Closing files, base filename: " + fileName );
             }
             dataFile.close();
             dataFile = null;
             keyFile.close();
             keyFile = null;
         }
         catch ( IOException e )
         {
             log.error( logCacheName + "Failure closing files in dispose, filename: " + fileName, e );
         }
 
         if ( log.isInfoEnabled() )
         {
             log.info( logCacheName + "Shutdown complete." );
         }
     }
 
     /**
      * Add descriptor to recycle bin if it is not null. Adds the length of the item to the bytes
      * free.
      * <p>
      * This is called in three places: (1) When an item is removed. All item removals funnel down to
      * the removeSingleItem method. (2) When an item on disk is updated with a value that will not
      * fit in the previous slot. (3) When the max key size is reached, the freed slot will be added.
      * <p>
      * The recylebin is not a set. If a slot it added twice, it will result in the wrong data being
      * returned.
      * <p>
      * @param ded
      */
     protected void addToRecycleBin( IndexedDiskElementDescriptor ded )
     {
         // reuse the spot
         if ( ded != null )
         {
             this.adjustBytesFree( ded, true );
 
             if ( doRecycle )
             {
                 recycle.add( ded );
                 if ( log.isDebugEnabled() )
                 {
                     log.debug( logCacheName + "recycled ded" + ded );
                 }
 
             }
         }
     }
 
     /**
      * Performs the check for optimization, and if it is required, do it.
      */
     protected void doOptimizeRealTime()
     {
         if ( isRealTimeOptimizationEnabled && !isOptimizing && ( removeCount++ >= cattr.getOptimizeAtRemoveCount() ) )
         {
             isOptimizing = true;
 
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "Optimizing file. removeCount [" + removeCount + "] OptimizeAtRemoveCount ["
                     + cattr.getOptimizeAtRemoveCount() + "]" );
             }
 
             if ( currentOptimizationThread == null )
             {
                 storageLock.writeLock().lock();
 
                 try
                 {
                     if ( currentOptimizationThread == null )
                     {
                         currentOptimizationThread = new Thread( new Runnable()
                         {
                             public void run()
                             {
                                 optimizeFile();
 
                                 currentOptimizationThread = null;
                             }
                         }, "IndexedDiskCache-OptimizationThread" );
                     }
                 }
                 finally
                 {
                     storageLock.writeLock().unlock();
                 }
 
                 if ( currentOptimizationThread != null )
                 {
                     currentOptimizationThread.start();
                 }
             }
         }
     }
 
     /**
      * File optimization is handled by this method. It works as follows:
      * <ol>
      * <li>Shutdown recycling and turn on queuing of puts. </li> <li>Take a snapshot of the current
      * descriptors. If there are any removes, ignore them, as they will be compacted during the next
      * optimization.</li> <li>Optimize the snapshot. For each descriptor:
      * <ol>
      * <li>Obtain the write-lock.</li> <li>Shift the element on the disk, in order to compact out
      * the free space. </li> <li>Release the write-lock. This allows elements to still be accessible
      * during optimization.</li>
      * </ol>
      * </li> <li>Obtain the write-lock.</li> <li>All queued puts are made at the end of the file.
      * Optimize these under a single write-lock.</li> <li>Truncate the file.</li> <li>Release the
      * write-lock. </li> <li>Restore system to standard operation.</li>
      * </ol>
      */
     protected void optimizeFile()
     {
         ElapsedTimer timer = new ElapsedTimer();
         timesOptimized++;
         if ( log.isInfoEnabled() )
         {
             log.info( logCacheName + "Beginning Optimization #" + timesOptimized );
         }
 
         // CREATE SNAPSHOT
         IndexedDiskElementDescriptor[] defragList = null;
 
         storageLock.writeLock().lock();
 
         try
         {
             queueInput = true;
             // shut off recycle while we're optimizing,
             doRecycle = false;
             defragList = createPositionSortedDescriptorList();
         }
         finally
         {
             // Release if I acquired.
             storageLock.writeLock().unlock();
         }
 
         // Defrag the file outside of the write lock. This allows a move to be made,
         // and yet have the element still accessible for reading or writing.
         long expectedNextPos = defragFile( defragList, 0 );
 
         // ADD THE QUEUED ITEMS to the end and then truncate
         storageLock.writeLock().lock();
 
         try
         {
             try
             {
                 if ( !queuedPutList.isEmpty() )
                 {
                     // This is perhaps unnecessary, but the list might not be as sorted as we think.
                     defragList = new IndexedDiskElementDescriptor[queuedPutList.size()];
                     queuedPutList.toArray( defragList );
                     Arrays.sort( defragList, new PositionComparator() );
 
                     // pack them at the end
                     expectedNextPos = defragFile( defragList, expectedNextPos );
                 }
                 // TRUNCATE THE FILE
                 dataFile.truncate( expectedNextPos );
             }
             catch ( Exception e )
             {
                 log.error( logCacheName + "Error optimizing queued puts.", e );
             }
 
             // RESTORE NORMAL OPERATION
             removeCount = 0;
             resetBytesFree();
             initializeRecycleBin();
             queuedPutList.clear();
             queueInput = false;
             // turn recycle back on.
             doRecycle = true;
             isOptimizing = false;
         }
         finally
         {
             storageLock.writeLock().unlock();
         }
 
         if ( log.isInfoEnabled() )
         {
             log.info( logCacheName + "Finished #" + timesOptimized + " Optimization took "
                 + timer.getElapsedTimeString() );
         }
     }
 
     /**
      * Defragments the file in place by compacting out the free space (i.e., moving records
      * forward). If there were no gaps the resulting file would be the same size as the previous
      * file. This must be supplied an ordered defragList.
      * <p>
      * @param defragList sorted list of descriptors for optimization
      * @param startingPos the start position in the file
      * @return this is the potential new file end
      */
     private long defragFile( IndexedDiskElementDescriptor[] defragList, long startingPos )
     {
         ElapsedTimer timer = new ElapsedTimer();
         long preFileSize = 0;
         long postFileSize = 0;
         long expectedNextPos = 0;
         try
         {
             preFileSize = this.dataFile.length();
             // find the first gap in the disk and start defragging.
             expectedNextPos = startingPos;
             for ( int i = 0; i < defragList.length; i++ )
             {
                 storageLock.writeLock().lock();
                 try
                 {
                     if ( expectedNextPos != defragList[i].pos )
                     {
                         dataFile.move( defragList[i], expectedNextPos );
                     }
                     expectedNextPos = defragList[i].pos + IndexedDisk.HEADER_SIZE_BYTES + defragList[i].len;
                 }
                 finally
                 {
                     storageLock.writeLock().unlock();
                 }
             }
 
             postFileSize = this.dataFile.length();
 
             // this is the potential new file end
             return expectedNextPos;
         }
         catch ( IOException e )
         {
             log.error( logCacheName + "Error occurred during defragmentation.", e );
         }
         finally
         {
             if ( log.isInfoEnabled() )
             {
                 log.info( logCacheName + "Defragmentation took " + timer.getElapsedTimeString()
                     + ". File Size (before=" + preFileSize + ") (after=" + postFileSize + ") (truncating to "
                     + expectedNextPos + ")" );
             }
         }
 
         return 0;
     }
 
     /**
      * Creates a snapshot of the IndexedDiskElementDescriptors in the keyHash and returns them
      * sorted by position in the dataFile.
      * <p>
      * TODO fix values() method on the LRU map.
      * <p>
      * @return IndexedDiskElementDescriptor[]
      */
     private IndexedDiskElementDescriptor[] createPositionSortedDescriptorList()
     {
         IndexedDiskElementDescriptor[] defragList = new IndexedDiskElementDescriptor[keyHash.size()];
         Iterator<Map.Entry<K, IndexedDiskElementDescriptor>> iterator = keyHash.entrySet().iterator();
         for ( int i = 0; iterator.hasNext(); i++ )
         {
             Map.Entry<K, IndexedDiskElementDescriptor> next = iterator.next();
             defragList[i] = next.getValue();
         }
 
         Arrays.sort( defragList, new PositionComparator() );
 
         return defragList;
     }
 
     /**
      * Returns the current cache size.
      * <p>
      * @return The size value
      */
     @Override
     public int getSize()
     {
         return keyHash.size();
     }
 
     /**
      * Returns the size of the recyclebin in number of elements.
      * <p>
      * @return The number of items in the bin.
      */
     protected int getRecyleBinSize()
     {
         return this.recycle.size();
     }
 
     /**
      * Returns the number of times we have used spots from the recycle bin.
      * <p>
      * @return The number of spots used.
      */
     protected int getRecyleCount()
     {
         return this.recycleCnt;
     }
 
     /**
      * Returns the number of bytes that are free. When an item is removed, its length is recorded.
      * When a spot is used form the recycle bin, the length of the item stored is recorded.
      * <p>
      * @return The number bytes free on the disk file.
      */
     protected synchronized long getBytesFree()
     {
         return this.bytesFree;
     }
 
     /**
      * Resets the number of bytes that are free.
      */
     private synchronized void resetBytesFree()
     {
         this.bytesFree = 0;
     }
 
     /**
      * To subtract you can pass in false for add..
      * <p>
      * @param ded
      * @param add
      */
     private synchronized void adjustBytesFree( IndexedDiskElementDescriptor ded, boolean add )
     {
         if ( ded != null )
         {
             int amount = ded.len + IndexedDisk.HEADER_SIZE_BYTES;
 
             if ( add )
             {
                 this.bytesFree += amount;
             }
             else
             {
                 this.bytesFree -= amount;
             }
         }
     }
 
     /**
      * This is for debugging and testing.
      * <p>
      * @return the length of the data file.
      * @throws IOException
      */
     protected long getDataFileSize()
         throws IOException
     {
         long size = 0;
 
         storageLock.readLock().lock();
 
         try
         {
             if ( dataFile != null )
             {
                 size = dataFile.length();
             }
         }
         finally
         {
             storageLock.readLock().unlock();
         }
 
         return size;
     }
 
     /**
      * For debugging. This dumps the values by default.
      */
     public void dump()
     {
         dump( true );
     }
 
     /**
      * For debugging.
      * <p>
      * @param dumpValues A boolean indicating if values should be dumped.
      */
     public void dump( boolean dumpValues )
     {
         if ( log.isDebugEnabled() )
         {
             log.debug( logCacheName + "[dump] Number of keys: " + keyHash.size() );
 
             for (Map.Entry<K, IndexedDiskElementDescriptor> e : keyHash.entrySet())
             {
                 K key = e.getKey();
                 IndexedDiskElementDescriptor ded = e.getValue();
 
                 log.debug( logCacheName + "[dump] Disk element, key: " + key + ", pos: " + ded.pos + ", ded.len"
                     + ded.len + ( ( dumpValues ) ? ( ", val: " + get( key ) ) : "" ) );
             }
         }
     }
 
     /**
      * @return Returns the AuxiliaryCacheAttributes.
      */
     public AuxiliaryCacheAttributes getAuxiliaryCacheAttributes()
     {
         return this.cattr;
     }
 
     /**
      * Gets basic stats for the disk cache.
      * <p>
      * @return String
      */
     @Override
     public String getStats()
     {
         return getStatistics().toString();
     }
 
     /**
      * Returns info about the disk cache.
      * <p>
      * (non-Javadoc)
      * @see org.apache.jcs.auxiliary.AuxiliaryCache#getStatistics()
      */
     @Override
     public synchronized IStats getStatistics()
     {
         IStats stats = new Stats();
         stats.setTypeName( "Indexed Disk Cache" );
 
         ArrayList<IStatElement> elems = new ArrayList<IStatElement>();
 
         IStatElement se = null;
 
         se = new StatElement();
         se.setName( "Is Alive" );
         se.setData( "" + alive );
         elems.add( se );
 
         se = new StatElement();
         se.setName( "Key Map Size" );
         if ( this.keyHash != null )
         {
             se.setData( "" + this.keyHash.size() );
         }
         else
         {
             se.setData( "-1" );
         }
         elems.add( se );
 
         try
         {
             se = new StatElement();
             se.setName( "Data File Length" );
             if ( this.dataFile != null )
             {
                 se.setData( "" + this.dataFile.length() );
             }
             else
             {
                 se.setData( "-1" );
             }
             elems.add( se );
         }
         catch ( Exception e )
         {
             log.error( e );
         }
 
         se = new StatElement();
         se.setName( "Hit Count" );
         se.setData( "" + this.hitCount.get() );
         elems.add( se );
 
         se = new StatElement();
         se.setName( "Bytes Free" );
         se.setData( "" + this.bytesFree );
         elems.add( se );
 
         se = new StatElement();
         se.setName( "Optimize Operation Count" );
         se.setData( "" + this.removeCount );
         elems.add( se );
 
         se = new StatElement();
         se.setName( "Times Optimized" );
         se.setData( "" + this.timesOptimized );
         elems.add( se );
 
         se = new StatElement();
         se.setName( "Recycle Count" );
         se.setData( "" + this.recycleCnt );
         elems.add( se );
 
         se = new StatElement();
         se.setName( "Recycle Bin Size" );
         se.setData( "" + this.recycle.size() );
         elems.add( se );
 
         se = new StatElement();
         se.setName( "Startup Size" );
         se.setData( "" + this.startupSize );
         elems.add( se );
 
         // get the stats from the super too
         // get as array, convert to list, add list to our outer list
         IStats sStats = super.getStatistics();
         IStatElement[] sSEs = sStats.getStatElements();
         List<IStatElement> sL = Arrays.asList( sSEs );
         elems.addAll( sL );
 
         // get an array and put them in the Stats object
         IStatElement[] ses = elems.toArray( new StatElement[0] );
         stats.setStatElements( ses );
 
         return stats;
     }
 
     /**
      * This is exposed for testing.
      * <p>
      * @return Returns the timesOptimized.
      */
     protected int getTimesOptimized()
     {
         return timesOptimized;
     }
 
     /**
      * This is used by the event logging.
      * <p>
      * @return the location of the disk, either path or ip.
      */
     @Override
     protected String getDiskLocation()
     {
         return dataFile.getFilePath();
     }
 
     /**
      * Compares IndexedDiskElementDescriptor based on their position.
      * <p>
      */
     protected static final class PositionComparator
         implements Comparator<IndexedDiskElementDescriptor>, Serializable
     {
         /** serialVersionUID */
         private static final long serialVersionUID = -8387365338590814113L;
 
         /**
          * Compares two descriptors based on position.
          * <p>
          * @see java.util.Comparator#compare(java.lang.Object, java.lang.Object)
          */
         public int compare( IndexedDiskElementDescriptor o1, IndexedDiskElementDescriptor o2 )
         {
             IndexedDiskElementDescriptor ded1 = o1;
             IndexedDiskElementDescriptor ded2 = o2;
 
             if ( ded1.pos < ded2.pos )
             {
                 return -1;
             }
             else if ( ded1.pos == ded2.pos )
             {
                 return 0;
             }
             else
             {
                 return 1;
             }
         }
     }
 
     /**
      * Class for recycling and lru. This implements the LRU overflow callback, so we can add items
      * to the recycle bin.
      */
     public class LRUMap
         extends LRUMapJCS<K, IndexedDiskElementDescriptor>
         // implements Serializable
     {
         /** Don't change */
         private static final long serialVersionUID = 4955079991472142198L;
 
         /**
          * <code>tag</code> tells us which map we are working on.
          */
         public String tag = "orig";
 
         /**
          * Default
          */
         public LRUMap()
         {
             super();
         }
 
         /**
          * @param maxKeySize
          */
         public LRUMap( int maxKeySize )
         {
             super( maxKeySize );
         }
 
         /**
          * This is called when the may key size is reached. The least recently used item will be
          * passed here. We will store the position and size of the spot on disk in the recycle bin.
          * <p>
          * @param key
          * @param value
          */
         @Override
         protected void processRemovedLRU(K key, IndexedDiskElementDescriptor value )
         {
             addToRecycleBin( value );
             if ( log.isDebugEnabled() )
             {
                 log.debug( logCacheName + "Removing key: [" + key + "] from key store." );
                 log.debug( logCacheName + "Key store size: [" + this.size() + "]." );
             }
 
             doOptimizeRealTime();
         }
     }
 }