protected static final Logger LOG = Logger.getLogger(HTreeMap.class.getName());

protected static final int BUCKET_OVERFLOW = 4;

protected static final int DIV8 = 3;

protected static final int MOD8 = 0x7;

/** is this a Map or Set? if false, entries do not have values, only keys are allowed*/

protected final boolean hasValues;

/**

protected final int hashSalt;

protected final Atomic.Long counter;

protected final Serializer<K> keySerializer;

protected final Serializer<V> valueSerializer;

protected final Hasher<K> hasher;

protected final Engine engine;

protected final boolean expireFlag;

protected final long expireTimeStart;

protected final long expire;

protected final boolean expireAccessFlag;

protected final long expireAccess;

protected final long expireMaxSize;

protected final long expireStoreSize;

protected final boolean expireMaxSizeFlag;

protected final long[] expireHeads;

protected final long[] expireTails;

protected final Fun.Function1<V,K> valueCreator;

protected final CountDownLatch closeLatch = new CountDownLatch(2);

protected final Runnable closeListener = new Runnable() {

@Override public void run() {

if(closeLatch.getCount()>1){

closeLatch.countDown();

}

try {

closeLatch.await();

} catch (InterruptedException e) {

throw new RuntimeException(e);

}

HTreeMap.this.engine.closeListenerUnregister(HTreeMap.this.closeListener);

}

};

/** node which holds key-value pair */

protected static final class LinkedNode<K,V>{

public final long next;

public final long expireLinkNodeRecid;

public final K key;

public final V value;

public LinkedNode(final long next, long expireLinkNodeRecid, final K key, final V value ){

this.key = key;

this.expireLinkNodeRecid = expireLinkNodeRecid;

this.value = value;

this.next = next;

}

}

protected final Serializer<LinkedNode<K,V>> LN_SERIALIZER = new Serializer<LinkedNode<K,V>>() {

@Override

public void serialize(DataOutput out, LinkedNode<K,V> value) throws IOException {

DataOutput2.packLong(out, value.next);

if(expireFlag)

DataOutput2.packLong(out, value.expireLinkNodeRecid);

keySerializer.serialize(out,value.key);

if(hasValues)

valueSerializer.serialize(out,value.value);

}

@Override

public LinkedNode<K,V> deserialize(DataInput in, int available) throws IOException {

assert(available!=0);

return new LinkedNode<K, V>(

DataInput2.unpackLong(in),

expireFlag?DataInput2.unpackLong(in):0L,

keySerializer.deserialize(in,-1),

hasValues? valueSerializer.deserialize(in,-1) : (V) BTreeMap.EMPTY

);

}

@Override

public int fixedSize() {

return -1;

}

};

protected static final Serializer<long[][]>DIR_SERIALIZER = new Serializer<long[][]>() {

@Override

public void serialize(DataOutput out, long[][] value) throws IOException {

assert(value.length==16);

//first write mask which indicate subarray nullability

int nulls = 0;

for(int i = 0;i<16;i++){

if(value[i]!=null){

for(long l:value[i]){

if(l!=0){

nulls |= 1<<i;

break;

}

}

}

}

out.writeShort(nulls);

//write non null subarrays

for(int i = 0;i<16;i++){

if(value[i]!=null){

assert(value[i].length==8);

for(long l:value[i]){

DataOutput2.packLong(out, l);

}

}

}

}

@Override

public long[][] deserialize(DataInput in, int available) throws IOException {

final long[][] ret = new long[16][];

//there are 16 subarrays, each bite indicates if subarray is null

int nulls = in.readUnsignedShort();

for(int i=0;i<16;i++){

if((nulls & 1)!=0){

final long[] subarray = new long[8];

for(int j=0;j<8;j++){

subarray[j] = DataInput2.unpackLong(in);

}

ret[i] = subarray;

}

nulls = nulls>>>1;

}

return ret;

}

@Override

public int fixedSize() {

return -1;

}

};

/** list of segments, this is immutable*/

protected final long[] segmentRecids;

protected final ReentrantReadWriteLock[] segmentLocks = new ReentrantReadWriteLock[16];

{

for(int i=0;i< 16;i++) segmentLocks[i]=new ReentrantReadWriteLock(CC.FAIR_LOCKS);

}

/**

public HTreeMap(Engine engine, long counterRecid, int hashSalt, long[] segmentRecids,

Serializer<K> keySerializer, Serializer<V> valueSerializer,

long expireTimeStart, long expire, long expireAccess, long expireMaxSize, long expireStoreSize,

long[] expireHeads, long[] expireTails, Fun.Function1<V, K> valueCreator,

Hasher hasher, boolean disableLocks) {

if(counterRecid<0) throw new IllegalArgumentException();

if(engine==null) throw new NullPointerException();

if(segmentRecids==null) throw new NullPointerException();

if(keySerializer==null) throw new NullPointerException();

SerializerBase.assertSerializable(keySerializer);

this.hasValues = valueSerializer!=null;

if(hasValues) {

SerializerBase.assertSerializable(valueSerializer);

}

if(segmentRecids.length!=16) throw new IllegalArgumentException();

this.engine = engine;

this.hashSalt = hashSalt;

this.segmentRecids = Arrays.copyOf(segmentRecids,16);

this.keySerializer = keySerializer;

this.valueSerializer = valueSerializer;

this.hasher = hasher!=null ? hasher : Hasher.BASIC;

if(expire==0 && expireAccess!=0){

expire = expireAccess;

}

if(expireMaxSize!=0 && counterRecid==0){

throw new IllegalArgumentException("expireMaxSize must have counter enabled");

}

this.expireFlag = expire !=0L || expireAccess!=0L || expireMaxSize!=0 || expireStoreSize!=0;

this.expire = expire;

this.expireTimeStart = expireTimeStart;

this.expireAccessFlag = expireAccess !=0L || expireMaxSize!=0 || expireStoreSize!=0;

this.expireAccess = expireAccess;

this.expireHeads = expireHeads==null? null : Arrays.copyOf(expireHeads,16);

this.expireTails = expireTails==null? null : Arrays.copyOf(expireTails,16);

this.expireMaxSizeFlag = expireMaxSize!=0;

this.expireMaxSize = expireMaxSize;

this.expireStoreSize = expireStoreSize;

this.valueCreator = valueCreator;

if(counterRecid!=0){

this.counter = new Atomic.Long(engine,counterRecid);

Bind.size(this,counter);

}else{

this.counter = null;

}

if(expireFlag){

Thread t = new Thread(new ExpireRunnable(this),"HTreeMap expirator");

t.setDaemon(true);

t.start();

engine.closeListenerRegister(closeListener);

}

}

protected static long[] preallocateSegments(Engine engine){

//prealocate segmentRecids, so we dont have to lock on those latter

long[] ret = new long[16];

for(int i=0;i<16;i++)

ret[i] = engine.put(new long[16][], DIR_SERIALIZER);

return ret;

}

@Override

public boolean containsKey(final Object o){

return getPeek(o)!=null;

}

@Override

public int size() {

long size = sizeLong();

if(size>Integer.MAX_VALUE) return Integer.MAX_VALUE;

return (int) size;

}

@Override

public long sizeLong() {

if(counter!=null)

return counter.get();

long counter = 0;

//search tree, until we find first non null

for(int i=0;i<16;i++){

try{

segmentLocks[i].readLock().lock();

final long dirRecid = segmentRecids[i];

counter+=recursiveDirCount(dirRecid);

}finally {

segmentLocks[i].readLock().unlock();

}

}

return counter;

}

private long recursiveDirCount(final long dirRecid) {

long[][] dir = engine.get(dirRecid, DIR_SERIALIZER);

long counter = 0;

for(long[] subdir:dir){

if(subdir == null) continue;

for(long recid:subdir){

if(recid == 0) continue;

if((recid&1)==0){

//reference to another subdir

recid = recid>>>1;

counter += recursiveDirCount(recid);

}else{

//reference to linked list, count it

recid = recid>>>1;

while(recid!=0){

LinkedNode n = engine.get(recid, LN_SERIALIZER);

if(n!=null){

counter++;

recid = n.next;

}else{

recid = 0;

}

}

}

}

}

return counter;

}

@Override

public boolean isEmpty() {

//search tree, until we find first non null

for(int i=0;i<16;i++){

try{

segmentLocks[i].readLock().lock();

long dirRecid = segmentRecids[i];

long[][] dir = engine.get(dirRecid, DIR_SERIALIZER);

for(long[] d:dir){

if(d!=null) return false;

}

}finally {

segmentLocks[i].readLock().unlock();

}

}

return true;

}

@Override

public V get(final Object o){

if(o==null) return null;

final int h = hash(o);

final int segment = h >>>28;

final Lock lock = expireAccessFlag ? segmentLocks[segment].writeLock() : segmentLocks[segment].readLock();

lock.lock();

LinkedNode<K,V> ln;

try{

ln = getInner(o, h, segment);

if(ln==null) return null;

if(expireAccessFlag) expireLinkBump(segment,ln.expireLinkNodeRecid,true);

}finally {

lock.unlock();

}

if(valueCreator==null|| ln.value!=null) return ln.value;

V value = valueCreator.run((K) o);

V prevVal = putIfAbsent((K) o,value);

if(prevVal!=null) return prevVal;

return value;

}

/**

public V getPeek(final Object key){

if(key==null) return null;

final int h = hash(key);

final int segment = h >>>28;

final Lock lock = segmentLocks[segment].readLock();

lock.lock();

try{

LinkedNode<K,V> ln = getInner(key, h, segment);

if(ln==null) return null;

return ln.value;

}finally {

lock.unlock();

}

}

protected LinkedNode<K,V> getInner(Object o, int h, int segment) {

long recid = segmentRecids[segment];

for(int level=3;level>=0;level--){

long[][] dir = engine.get(recid, DIR_SERIALIZER);

if(dir == null) return null;

int slot = (h>>>(level*7 )) & 0x7F;

assert(slot<128);

if(dir[slot>>>DIV8]==null) return null;

recid = dir[slot>>>DIV8][slot&MOD8];

if(recid == 0) return null;

if((recid&1)!=0){ //last bite indicates if referenced record is LinkedNode

recid = recid>>>1;

while(true){

LinkedNode<K,V> ln = engine.get(recid, LN_SERIALIZER);

if(ln == null) return null;

if(hasher.equals(ln.key, (K) o)){

assert(hash(ln.key)==h);

return ln;

}

if(ln.next==0) return null;

recid = ln.next;

}

}

recid = recid>>>1;

}

return null;

}

@Override

public V put(final K key, final V value){

if (key == null)

throw new IllegalArgumentException("null key");

if (value == null)

throw new IllegalArgumentException("null value");

final int h = hash(key);

final int segment = h >>>28;

segmentLocks[segment].writeLock().lock();

try{

return putInner(key, value, h, segment);

}finally {

segmentLocks[segment].writeLock().unlock();

}

}

private V putInner(K key, V value, int h, int segment) {

long dirRecid = segmentRecids[segment];

int level = 3;

while(true){

long[][] dir = engine.get(dirRecid, DIR_SERIALIZER);

final int slot = (h>>>(7*level )) & 0x7F;

assert(slot<=127);

if(dir == null ){

//create new dir

dir = new long[16][];

}

if(dir[slot>>>DIV8] == null){

dir = Arrays.copyOf(dir, 16);

dir[slot>>>DIV8] = new long[8];

}

int counter = 0;

long recid = dir[slot>>>DIV8][slot&MOD8];

if(recid!=0){

if((recid&1) == 0){

dirRecid = recid>>>1;

level--;

continue;

}

recid = recid>>>1;

//traverse linked list, try to replace previous value

LinkedNode<K,V> ln = engine.get(recid, LN_SERIALIZER);

while(ln!=null){

if(hasher.equals(ln.key,key)){

//found, replace value at this node

V oldVal = ln.value;

ln = new LinkedNode<K, V>(ln.next, ln.expireLinkNodeRecid, ln.key, value);

engine.update(recid, ln, LN_SERIALIZER);

if(expireFlag) expireLinkBump(segment,ln.expireLinkNodeRecid,false);

notify(key, oldVal, value);

return oldVal;

}

recid = ln.next;

ln = recid==0? null : engine.get(recid, LN_SERIALIZER);

counter++;

}

//key was not found at linked list, so just append it to beginning

}

//check if linked list has overflow and needs to be expanded to new dir level

if(counter>=BUCKET_OVERFLOW && level>=1){

long[][] nextDir = new long[16][];

{

final long expireNodeRecid = expireFlag? engine.preallocate():0L;

final LinkedNode<K,V> node = new LinkedNode<K, V>(0, expireNodeRecid, key, value);

final long newRecid = engine.put(node, LN_SERIALIZER);

//add newly inserted record

int pos =(h >>>(7*(level-1) )) & 0x7F;

nextDir[pos>>>DIV8] = new long[8];

nextDir[pos>>>DIV8][pos&MOD8] = ( newRecid<<1) | 1;

if(expireFlag) expireLinkAdd(segment,expireNodeRecid,newRecid,h);

}

//redistribute linked bucket into new dir

long nodeRecid = dir[slot>>>DIV8][slot&MOD8]>>>1;

while(nodeRecid!=0){

LinkedNode<K,V> n = engine.get(nodeRecid, LN_SERIALIZER);

final long nextRecid = n.next;

int pos = (hash(n.key) >>>(7*(level -1) )) & 0x7F;

if(nextDir[pos>>>DIV8]==null) nextDir[pos>>>DIV8] = new long[8];

n = new LinkedNode<K, V>(nextDir[pos>>>DIV8][pos&MOD8]>>>1, n.expireLinkNodeRecid, n.key, n.value);

nextDir[pos>>>DIV8][pos&MOD8] = (nodeRecid<<1) | 1;

engine.update(nodeRecid, n, LN_SERIALIZER);

nodeRecid = nextRecid;

}

//insert nextDir and update parent dir

long nextDirRecid = engine.put(nextDir, DIR_SERIALIZER);

int parentPos = (h>>>(7*level )) & 0x7F;

dir = Arrays.copyOf(dir,16);

dir[parentPos>>>DIV8] = Arrays.copyOf(dir[parentPos>>>DIV8],8);

dir[parentPos>>>DIV8][parentPos&MOD8] = (nextDirRecid<<1) | 0;

engine.update(dirRecid, dir, DIR_SERIALIZER);

notify(key, null, value);

return null;

}else{

// record does not exist in linked list, so create new one

recid = dir[slot>>>DIV8][slot&MOD8]>>>1;

final long expireNodeRecid = expireFlag? engine.put(ExpireLinkNode.EMPTY, ExpireLinkNode.SERIALIZER):0L;

final long newRecid = engine.put(new LinkedNode<K, V>(recid, expireNodeRecid, key, value), LN_SERIALIZER);

dir = Arrays.copyOf(dir,16);

dir[slot>>>DIV8] = Arrays.copyOf(dir[slot>>>DIV8],8);

dir[slot>>>DIV8][slot&MOD8] = (newRecid<<1) | 1;

engine.update(dirRecid, dir, DIR_SERIALIZER);

if(expireFlag) expireLinkAdd(segment,expireNodeRecid, newRecid,h);

notify(key, null, value);

return null;

}

}

}

@Override

public V remove(Object key){

final int h = hash(key);

final int segment = h >>>28;

segmentLocks[segment].writeLock().lock();

try{

return removeInternal(key, segment, h, true);

}finally {

segmentLocks[segment].writeLock().unlock();

}

}

protected V removeInternal(Object key, int segment, int h, boolean removeExpire){

final long[] dirRecids = new long[4];

int level = 3;

dirRecids[level] = segmentRecids[segment];

assert(segment==h>>>28);

while(true){

long[][] dir = engine.get(dirRecids[level], DIR_SERIALIZER);

final int slot = (h>>>(7*level )) & 0x7F;

assert(slot<=127);

if(dir == null ){

//create new dir

dir = new long[16][];

}

if(dir[slot>>>DIV8] == null){

dir = Arrays.copyOf(dir,16);

dir[slot>>>DIV8] = new long[8];

}

long recid = dir[slot>>>DIV8][slot&MOD8];

if(recid!=0){

if((recid&1) == 0){

level--;

dirRecids[level] = recid>>>1;

continue;

}

recid = recid>>>1;

//traverse linked list, try to remove node

LinkedNode<K,V> ln = engine.get(recid, LN_SERIALIZER);

LinkedNode<K,V> prevLn = null;

long prevRecid = 0;

while(ln!=null){

if(hasher.equals(ln.key, (K) key)){

//remove from linkedList

if(prevLn == null ){

//referenced directly from dir

if(ln.next==0){

recursiveDirDelete(h, level, dirRecids, dir, slot);

}else{

dir=Arrays.copyOf(dir,16);

dir[slot>>>DIV8] = Arrays.copyOf(dir[slot>>>DIV8],8);

dir[slot>>>DIV8][slot&MOD8] = (ln.next<<1)|1;

engine.update(dirRecids[level], dir, DIR_SERIALIZER);

}

}else{

//referenced from LinkedNode

prevLn = new LinkedNode<K, V>(ln.next, prevLn.expireLinkNodeRecid,prevLn.key, prevLn.value);

engine.update(prevRecid, prevLn, LN_SERIALIZER);

}

//found, remove this node

assert(hash(ln.key)==h);

engine.delete(recid, LN_SERIALIZER);

if(removeExpire && expireFlag) expireLinkRemove(segment, ln.expireLinkNodeRecid);

notify((K) key, ln.value, null);

return ln.value;

}

prevRecid = recid;

prevLn = ln;

recid = ln.next;

ln = recid==0? null : engine.get(recid, LN_SERIALIZER);

}

//key was not found at linked list, so it does not exist

return null;

}

//recid is 0, so entry does not exist

return null;

}

}

private void recursiveDirDelete(int h, int level, long[] dirRecids, long[][] dir, int slot) {

//was only item in linked list, so try to collapse the dir

dir=Arrays.copyOf(dir,16);

dir[slot>>>DIV8] = Arrays.copyOf(dir[slot>>>DIV8],8);

dir[slot>>>DIV8][slot&MOD8] = 0;

//one record was zeroed out, check if subarray can be collapsed to null

boolean allZero = true;

for(long l:dir[slot>>>DIV8]){

if(l!=0){

allZero = false;

break;

}

}

if(allZero){

dir[slot>>>DIV8] = null;

}

allZero = true;

for(long[] l:dir){

if(l!=null){

allZero = false;

break;

}

}

if(allZero){

//delete from parent dir

if(level==3){

//parent is segment, recid of this dir can not be modified, so just update to null

engine.update(dirRecids[level], new long[16][], DIR_SERIALIZER);

}else{

engine.delete(dirRecids[level], DIR_SERIALIZER);

final long[][] parentDir = engine.get(dirRecids[level + 1], DIR_SERIALIZER);

final int parentPos = (h >>> (7 * (level + 1))) & 0x7F;

recursiveDirDelete(h,level+1,dirRecids, parentDir, parentPos);

//parentDir[parentPos>>>DIV8][parentPos&MOD8] = 0;

//engine.update(dirRecids[level + 1],parentDir,DIR_SERIALIZER);

}

}else{

engine.update(dirRecids[level], dir, DIR_SERIALIZER);

}

}

@Override

public void clear() {

for(int i = 0; i<16;i++) try{

segmentLocks[i].writeLock().lock();

final long dirRecid = segmentRecids[i];

recursiveDirClear(dirRecid);

//set dir to null, as segment recid is immutable

engine.update(dirRecid, new long[16][], DIR_SERIALIZER);

if(expireFlag)

while(expireLinkRemoveLast(i)!=null){} //TODO speedup remove all

}finally {

segmentLocks[i].writeLock().unlock();

}

}

private void recursiveDirClear(final long dirRecid) {

final long[][] dir = engine.get(dirRecid, DIR_SERIALIZER);

if(dir == null) return;

for(long[] subdir:dir){

if(subdir==null) continue;

for(long recid:subdir){

if(recid == 0) continue;

if((recid&1)==0){

//another dir

recid = recid>>>1;

//recursively remove dir

recursiveDirClear(recid);

engine.delete(recid, DIR_SERIALIZER);

}else{

//linked list to delete

recid = recid>>>1;

while(recid!=0){

LinkedNode n = engine.get(recid, LN_SERIALIZER);

engine.delete(recid,LN_SERIALIZER);

notify((K)n.key, (V)n.value , null);

recid = n.next;

}

}

}

}

}

@Override

public boolean containsValue(Object value) {

for (V v : values()) {

if (v.equals(value)) return true;

}

return false;

}

protected class KeySet extends AbstractSet<K> {

@Override

public int size() {

return HTreeMap.this.size();

}

@Override

public boolean isEmpty() {

return HTreeMap.this.isEmpty();

}

@Override

public boolean contains(Object o) {

return HTreeMap.this.containsKey(o);

}

@Override

public Iterator<K> iterator() {

return new KeyIterator();

}

@Override

public boolean add(K k) {

if(HTreeMap.this.hasValues)

throw new UnsupportedOperationException();

else

return HTreeMap.this.put(k, (V) BTreeMap.EMPTY) == null;

}

@Override

public boolean remove(Object o) {

return HTreeMap.this.remove(o)!=null;

}

@Override

public void clear() {

HTreeMap.this.clear();

}

public HTreeMap<K,V> parent(){

return HTreeMap.this;

}

@Override

public int hashCode() {

int result = 0;

for (K k : this) {

result += hasher.hashCode(k);

}

return result;

}

}

private final Set<K> _keySet = new KeySet();

@Override

public Set<K> keySet() {

return _keySet;

}

private final Collection<V> _values = new AbstractCollection<V>(){

@Override

public int size() {

return HTreeMap.this.size();

}

@Override

public boolean isEmpty() {

return HTreeMap.this.isEmpty();

}

@Override

public boolean contains(Object o) {

return HTreeMap.this.containsValue(o);

}

@Override

public Iterator<V> iterator() {

return new ValueIterator();

}

};

@Override

public Collection<V> values() {

return _values;

}

private final Set<Entry<K,V>> _entrySet = new AbstractSet<Entry<K,V>>(){

@Override

public int size() {

return HTreeMap.this.size();

}

@Override

public boolean isEmpty() {

return HTreeMap.this.isEmpty();

}

@Override

public boolean contains(Object o) {

if(o instanceof Entry){

Entry e = (Entry) o;

Object val = HTreeMap.this.get(e.getKey());

return val!=null && val.equals(e.getValue());

}else

return false;

}

@Override

public Iterator<Entry<K, V>> iterator() {

return new EntryIterator();

}

@Override

public boolean add(Entry<K, V> kvEntry) {

K key = kvEntry.getKey();

V value = kvEntry.getValue();

if(key==null || value == null) throw new NullPointerException();

HTreeMap.this.put(key, value);

return true;

}

@Override

public boolean remove(Object o) {

if(o instanceof Entry){

Entry e = (Entry) o;

Object key = e.getKey();

if(key == null) return false;

return HTreeMap.this.remove(key, e.getValue());

}

return false;

}

@Override

public void clear() {

HTreeMap.this.clear();

}

};

@Override

public Set<Entry<K, V>> entrySet() {

return _entrySet;

}

protected int hash(final Object key) {

int h = hasher.hashCode((K) key) ^ hashSalt;

h ^= (h >>> 20) ^ (h >>> 12);

return h ^ (h >>> 7) ^ (h >>> 4);

}

abstract class HashIterator{

protected LinkedNode[] currentLinkedList;

protected int currentLinkedListPos = 0;

private K lastReturnedKey = null;

private int lastSegment = 0;

HashIterator(){

currentLinkedList = findNextLinkedNode(0);

}

public void remove() {

final K keyToRemove = lastReturnedKey;

if (lastReturnedKey == null)

throw new IllegalStateException();

lastReturnedKey = null;

HTreeMap.this.remove(keyToRemove);

}

public boolean hasNext(){

return currentLinkedList!=null && currentLinkedListPos<currentLinkedList.length;

}

protected void moveToNext(){

lastReturnedKey = (K) currentLinkedList[currentLinkedListPos].key;

currentLinkedListPos+=1;

if(currentLinkedListPos==currentLinkedList.length){

final int lastHash = hash(lastReturnedKey);

currentLinkedList = advance(lastHash);

currentLinkedListPos = 0;

}

}