/*
 *  Copyright (c) 2012 Jan Kotek
 *
 *  Licensed 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.mapdb;

/**
 * Fixed size cache which uses hash table.
 * Is thread-safe and does not require any locks.
 * Items are randomly removed and replaced by hash collisions.
 * <p/>
 * This is simple, concurrent, small-overhead, random cache.
 *
 * @author Jan Kotek
 */
public class CacheHashTable implements Engine {

    protected Engine engine;

    protected HashItem[] items;
    protected final int cacheMaxSize;

    private class HashItem {
        final long key;
        final Object val;

        private HashItem(long key, Object val) {
            this.key = key;
            this.val = val;
        }
    }



    public CacheHashTable(Engine engine, int cacheMaxSize) {
        this.engine = engine;
        this.items = new HashItem[cacheMaxSize];
        this.cacheMaxSize = cacheMaxSize;
    }

    @Override
    public <A> long recordPut(A value, Serializer<A> serializer) {
        final long recid = engine.recordPut(value, serializer);
        items[Math.abs(Utils.longHash(recid))%cacheMaxSize] = new HashItem(recid, value);
        return recid;
    }

    @Override
    @SuppressWarnings("unchecked")
    public <A> A recordGet(long recid, Serializer<A> serializer) {
        final int pos = Math.abs(Utils.longHash(recid))%cacheMaxSize;
        HashItem item = items[pos];
        if(item!=null && recid == item.key)
            return (A) item.val;

        //not in cache, fetch and add
        final A value = engine.recordGet(recid, serializer);
        if(value!=null)
            items[pos] = new HashItem(recid, value);
        return value;
    }

    @Override
    public <A> void recordUpdate(long recid, A value, Serializer<A> serializer) {
        items[Math.abs(Math.abs(Utils.longHash(recid)))%cacheMaxSize] = new HashItem(recid, value);
        engine.recordUpdate(recid, value, serializer);
    }

    @Override
    public void recordDelete(long recid) {
        final int pos = Math.abs(Utils.longHash(recid))%cacheMaxSize;
        HashItem item = items[pos];
        if(item!=null && recid == item.key)
            items[pos] = null;
    }


    @Override
    public void close() {
        engine.close();
        //dereference to prevent memory leaks
        engine = null;
        items = null;
    }

    @Override
    public void commit() {
        engine.commit();
    }

    @Override
    public void rollback() {
        for(int i = 0;i<items.length;i++)
            items[i] = null;
        engine.rollback();
    }

    @Override
    public long serializerRecid() {
        return engine.serializerRecid();
    }

    @Override
    public long nameDirRecid() {
        return engine.nameDirRecid();
    }

    @Override
    public boolean isReadOnly() {
        return engine.isReadOnly();
    }

}

@@ -619,16 +619,69 @@
 mapdb;%0A%0A
+import java.util.concurrent.atomic.AtomicReference;%0A%0A
 /**%0A * F
@@ -1022,24 +1022,40 @@
 otected 
+AtomicReference%3C
 HashItem
 %5B%5D items
@@ -1046,16 +1046,17 @@
 HashItem
+%3E
 %5B%5D items
@@ -1416,24 +1416,31 @@
 s = new 
-HashItem
+AtomicReference
 %5BcacheMa
@@ -1485,24 +1485,120 @@
 cheMaxSize;%0A
+        for(int i=0;i%3CcacheMaxSize;i++)%0A            items%5Bi%5D = new AtomicReference%3CHashItem%3E();%0A
     %7D%0A%0A    @
@@ -1737,38 +1737,48 @@
 lizer);%0A        
-items%5B
+final int pos = 
 Math.abs(Utils.l
@@ -1797,36 +1797,57 @@
 d))%25cacheMaxSize
-%5D = 
+;%0A        items%5Bpos%5D.set(
 new HashItem(rec
@@ -1848,32 +1848,33 @@
 em(recid, value)
+)
 ;%0A        return
@@ -2100,32 +2100,38 @@
 tem = items%5Bpos%5D
+.get()
 ;%0A        if(ite
@@ -2339,18 +2339,36 @@
 ems%5Bpos%5D
- =
+.compareAndSet(item,
  new Has
@@ -2378,32 +2378,33 @@
 em(recid, value)
+)
 ;%0A        return
@@ -2522,22 +2522,32 @@
         
-items%5B
+final int pos = 
 Math.abs
@@ -2596,37 +2596,53 @@
 Size
-%5D = new HashItem(recid, value
+;%0A        HashItem expected = items%5Bpos%5D.get(
 );%0A 
@@ -2691,24 +2691,96 @@
 erializer);%0A
+        items%5Bpos%5D.compareAndSet(expected, new HashItem(recid, value));%0A
     %7D%0A%0A    @
@@ -2910,36 +2910,40 @@
        HashItem 
-item
+expected
  = items%5Bpos%5D;%0A 
@@ -2939,16 +2939,22 @@
 ems%5Bpos%5D
+.get()
 ;%0A      
@@ -2947,39 +2947,79 @@
 .get();%0A        
-if(item
+engine.recordDelete(recid);%0A        if(expected
 !=null && recid 
@@ -3013,36 +3013,40 @@
 ull && recid == 
-item
+expected
 .key)%0A          
@@ -3057,23 +3057,45 @@
 ems%5Bpos%5D
- = 
+.compareAndSet(expected,
 null
+)
 ;%0A    %7D%0A
@@ -3436,23 +3436,26 @@
 items%5Bi%5D
- = 
+.set(
 null
+)
 ;%0A      

Fix issue #9, there was race condition in Cache