extends AbstractMap<K,V>

implements ConcurrentMap<K, V>,

Bind.MapWithModificationListener<K,V>,

Closeable {

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;

/** number of segments. Must be 16 in production, can be also 1 for debugging */

static final int SEG = 16;

/** 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 long[] counterRecids;

protected final Serializer<K> keySerializer;

protected final Serializer<V> valueSerializer;

protected final Engine[] engines;

protected final boolean closeEngine;

protected final boolean expireFlag;

protected final boolean expireSingleThreadFlag;

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 long[] expireStoreSizes;

protected final long[] expireStoreSizesCompact;

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

/**

protected final boolean closeExecutor;

protected final ScheduledExecutorService executor;

protected final Lock consistencyLock;

/** 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 ){

if(CC.ASSERT && next>>>48!=0)

throw new DBException.DataCorruption("next recid too big");

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>>() {

/** used to check that every 64000 th element has consistent has befor and after (de)serialization*/

int serCounter = 0;

@Override

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

if(((serCounter++ )& 0xFFFF)==0){

assertHashConsistent(value.key);

}

DataIO.packLong(out, value.next);

if(expireFlag)

DataIO.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 {

if(CC.ASSERT && available==0)

throw new AssertionError();

return new LinkedNode<K, V>(

DataIO.unpackLong(in),

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

keySerializer.deserialize(in,-1),

hasValues? valueSerializer.deserialize(in,-1) : (V) Boolean.TRUE

);

}

@Override

public boolean isTrusted() {

return keySerializer.isTrusted() && valueSerializer.isTrusted();

}

};

private final void assertHashConsistent(K key) throws IOException {

int hash = keySerializer.hashCode(key);

DataIO.DataOutputByteArray out = new DataIO.DataOutputByteArray();

keySerializer.serialize(out,key);

DataIO.DataInputByteArray in = new DataIO.DataInputByteArray(out.buf, 0);

K key2 = keySerializer.deserialize(in,-1);

if(hash!=keySerializer.hashCode(key2)){

throw new IllegalArgumentException("Key does not have consistent hash before and after deserialization. Class: "+key.getClass());

}

if(!keySerializer.equals(key,key2)){

throw new IllegalArgumentException("Key does not have consistent equals before and after deserialization. Class: "+key.getClass());

}

if(out.pos!=in.pos){

throw new IllegalArgumentException("Key has inconsistent serialization length. Class: "+key.getClass());

}

}

protected static final Serializer<Object> DIR_SERIALIZER = new Serializer<Object>() {

@Override

public void serialize(DataOutput out, Object value) throws IOException {

DataIO.DataOutputByteArray out2 = (DataIO.DataOutputByteArray) out;

if(value instanceof long[]) {

serializeLong(out2, value);

return;

}

int[] c = (int[]) value;

if(CC.ASSERT){

int len = 4 +

Integer.bitCount(c[0])+

Integer.bitCount(c[1])+

Integer.bitCount(c[2])+

Integer.bitCount(c[3]);

if(len!=c.length)

throw new DBException.DataCorruption("bitmap!=len");

}

//write bitmaps

out2.writeInt(c[0]);

out2.writeInt(c[1]);

out2.writeInt(c[2]);

out2.writeInt(c[3]);

if(c.length==4)

return;

out2.packLong((((long)c[4])<<1)|1L);

for(int i=5;i<c.length;i++){

out2.packLong(c[i]);

}

}

private void serializeLong(DataIO.DataOutputByteArray out, Object value) throws IOException {

long[] c= (long[]) value;

if(CC.ASSERT){

int len = 2 +

Long.bitCount(c[0])+

Long.bitCount(c[1]);

if(len!=c.length)

throw new DBException.DataCorruption("bitmap!=len");

}

out.writeLong(c[0]);

out.writeLong(c[1]);

if(c.length==2)

return;

out.packLong(c[2]<<1);

for(int i=3;i<c.length;i++){

out.packLong(c[i]);

}

}

@Override

public Object deserialize(DataInput in, int available) throws IOException {

DataIO.DataInputInternal in2 = (DataIO.DataInputInternal) in;

//length of dir is 128 longs, each long has 6 bytes (not 8)

//to save memory zero values are skipped,

//there is bitmap at first 16 bytes, each non-zero long has bit set

//to determine offset one must traverse bitmap and count number of bits set

int bitmap1 = in.readInt();

int bitmap2 = in.readInt();

int bitmap3 = in.readInt();

int bitmap4 = in.readInt();

int len = Integer.bitCount(bitmap1) + Integer.bitCount(bitmap2) + Integer.bitCount(bitmap3) + Integer.bitCount(bitmap4);

if (len == 0) {

return new int[4];

}

long firstVal = in2.unpackLong();

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

//return longs

long[] ret = new long[2 + len];

ret[0] = ((long) bitmap1 << 32) | (bitmap2 & 0xFFFFFFFF);

ret[1] = ((long) bitmap3 << 32) | (bitmap4 & 0xFFFFFFFF);

ret[2] = firstVal >>> 1;

len += 2;

in2.unpackLongArray(ret, 3, len);

return ret;

} else {

//return int[]

int[] ret = new int[4 + len];

ret[0] = bitmap1;

ret[1] = bitmap2;

ret[2] = bitmap3;

ret[3] = bitmap4;

ret[4] = (int) (firstVal >>> 1);

len += 4;

in2.unpackIntArray(ret,5,len);

return ret;

}

}

@Override

public boolean isTrusted() {

return true;

}

};

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

protected final long[] segmentRecids;

protected final ReentrantReadWriteLock[] segmentLocks;

/**

public HTreeMap(

Engine[] engines,

boolean closeEngine,

long[] counterRecids,

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,

ScheduledExecutorService executor,

long executorPeriod,

boolean closeExecutor,

Lock consistencyLock) {

if(counterRecids!=null && counterRecids.length!=SEG)

throw new IllegalArgumentException();

if(engines==null)

throw new NullPointerException();

if(engines.length!=SEG)

throw new IllegalArgumentException("engines wrong length");

if(segmentRecids==null)

throw new NullPointerException();

if(keySerializer==null)

throw new NullPointerException();

this.hasValues = valueSerializer!=null;

segmentLocks=new ReentrantReadWriteLock[SEG];

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

segmentLocks[i]=new ReentrantReadWriteLock(CC.FAIR_LOCKS);

}

this.closeEngine = closeEngine;

this.closeExecutor = closeExecutor;

this.engines = engines.clone();

this.hashSalt = hashSalt;

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

this.keySerializer = keySerializer;

this.valueSerializer = valueSerializer;

this.consistencyLock = consistencyLock ==null? Store.NOLOCK : consistencyLock;

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

expire = expireAccess;

}

if(expireMaxSize!=0 && counterRecids==null){

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,SEG);

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

this.expireMaxSizeFlag = expireMaxSize!=0;

this.expireMaxSize = expireMaxSize;

this.expireStoreSize = expireStoreSize;

this.valueCreator = valueCreator;

if(counterRecids!=null){

// use per-segment counter and sum all segments in map.size()

this.counterRecids = counterRecids.clone();

}else{

this.counterRecids = null;

}

expireSingleThreadFlag = (expireFlag && executor==null);

this.executor = executor;

if(expireStoreSize>0){

expireStoreSizesCompact = new long[engines.length];

expireStoreSizes = new long[engines.length];

for(int i=0;i<engines.length;i++) {

Engine e = engines[i];

boolean isFirstEngine = true;

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

if (engines[j] == e) {

isFirstEngine = false;

break;

}

}

long thisEngineCount = 0;

for (Engine e2 : engines) {

if (e2 == e)

thisEngineCount++;

}

expireStoreSizes[i] = (long) (0.85 * expireStoreSize * thisEngineCount / engines.length);

if (isFirstEngine){

expireStoreSizesCompact[i] = expireStoreSize * thisEngineCount / engines.length;

}else{

expireStoreSizesCompact[i] = 0;

}

}

}else{

expireStoreSizes = null;

expireStoreSizesCompact = null;

}

if(expireFlag && executor!=null){

if(executor!=null && engines[0].canRollback()) {

//TODO this should be covered by SequentialLock, check it, and remove warning

LOG.warning("HTreeMap Expiration should not be used with transaction enabled. It can lead to data corruption, commit might happen while background thread works, and only part of expiration data will be commited.");

}

//schedule expirators for all segments

for(int i=0;i<segmentLocks.length;i++) {

final int seg = i;

final Lock lock = segmentLocks[seg].writeLock();

executor.scheduleAtFixedRate(new Runnable() {

@Override

public void run() {

long removePerSegment = HTreeMap.this.expireCalcRemovePerSegment();

if(HTreeMap.this.expire==0 && HTreeMap.this.expireAccess==0 && removePerSegment<=0 )

return;

lock.lock();

try {

HTreeMap.this.expirePurgeSegment(seg, removePerSegment);

}finally{

lock.unlock();

}

}

},

(long) (executorPeriod * Math.random()),

executorPeriod,

TimeUnit.MILLISECONDS);

}

}

}

protected static long[] preallocateSegments(Engine[] engines){

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

long[] ret = new long[SEG];

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

ret[i] = engines[i].put(new int[4], DIR_SERIALIZER);

return ret;

}

@Override

public boolean containsKey(final Object o){

return getPeek(o)!=null;

}

@Override

public int size() {

return (int) Math.min(sizeLong(), Integer.MAX_VALUE);

}

@Override

public long sizeLong() {

if(counterRecids!=null) {

long ret = 0;

for(int i=0;i<counterRecids.length;i++){

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

lock.lock();

try {

ret += engines[i].get(counterRecids[i], Serializer.LONG);

}finally {

lock.unlock();

}

}

return ret;

}

long counter = 0;

//search tree, until we find first non null

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

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

lock.lock();

try{

final long dirRecid = segmentRecids[i];

counter+=recursiveDirCount(engines[i],dirRecid);

}finally {

lock.unlock();

}

}

return counter;

}

public long mappingCount(){

//method added in java 8

return sizeLong();

}

private long recursiveDirCount(Engine engine,final long dirRecid) {

Object dir = engine.get(dirRecid, DIR_SERIALIZER);

long counter = 0;

int dirLen = dirLen(dir);

for(int pos=dirStart(dir);pos<dirLen;pos++){

long recid = dirGet(dir,pos);

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

//reference to another subdir

recid = recid>>>1;

counter += recursiveDirCount(engine, 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() {

if(counterRecids!=null){

for(int i=0;i<counterRecids.length;i++){

if(0==engines[i].get(counterRecids[i],Serializer.LONG))

return true;

}

return false;

}

//search tree, until we find first non null

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

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

lock.lock();

try{

long dirRecid = segmentRecids[i];

Object dir = engines[i].get(dirRecid, DIR_SERIALIZER);

if(!dirIsEmpty(dir)){

return false;

}

}finally {

lock.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 && expireAccessFlag)

expireLinkBump(segment,ln.expireLinkNodeRecid,true); //TODO sequential lock here?

}finally {

lock.unlock();

}

if(expireSingleThreadFlag)

expirePurge();

if(valueCreator==null){

if(ln==null)

return null;

return ln.value;

}

//value creator is set, so create and put new value

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

//there is race condition, valueCreator could be called twice. But map will be updated only once

V prevVal = value==null ?

null :

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;

V ret;

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

lock.lock();

try{

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

ret = ln==null?

null:

ln.value;

}finally {

lock.unlock();

}

if(expireSingleThreadFlag)

expirePurge();

return ret;

}

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

long recid = segmentRecids[segment];

Engine engine = engines[segment];

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

Object dir = engine.get(recid, DIR_SERIALIZER);

if(dir == null)

return null;

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

if(CC.ASSERT && slot>128)

throw new DBException.DataCorruption("slot too high");

recid = dirGetSlot(dir, slot);

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(keySerializer.equals(ln.key, (K) o)){

if(CC.ASSERT && hash(ln.key)!=h)

throw new DBException.DataCorruption("inconsistent hash");

return ln;

}

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

recid = ln.next;

}

}

recid = recid>>>1;

}

return null;

}

protected static boolean dirIsEmpty(Object dir) {

if(dir == null)

return true;

if(dir instanceof long[])

return false;

return ((int[])dir).length==4;

}

protected static int dirLen(Object dir) {

return dir instanceof int[]?

((int[])dir).length:

((long[])dir).length;

}

protected static int dirStart(Object dir) {

return dir instanceof int[]?4:2;

}

protected static long dirGet(Object dir, int pos) {

return dir instanceof int[]?

((int[])dir)[pos]:

((long[])dir)[pos];

}

protected long dirGetSlot(Object dir, int slot) {

if(dir instanceof int[]){

int[] cc = (int[]) dir;

int pos = dirOffsetFromSlot(cc,slot);

if(pos<0)

return 0;

return cc[pos];

}else{

long[] cc = (long[]) dir;

int pos = dirOffsetFromSlot(cc,slot);

if(pos<0)

return 0;

return cc[pos];

}

}

protected static int dirOffsetFromSlot(Object dir, int slot) {

if(dir instanceof int[])

return dirOffsetFromSlot((int[])dir,slot);

else

return dirOffsetFromSlot((long[])dir,slot);

}

/** converts hash slot into actual offset in dir array, using bitmap */

protected static final int dirOffsetFromSlot(int[] dir, int slot) {

if(CC.ASSERT && slot>127)

throw new DBException.DataCorruption("slot too high");

int val = slot>>>5;

slot &=31;

int isSet = ((dir[val] >>> (slot)) & 1); //check if bit at given slot is set

isSet <<=1; //multiply by two, so it is usable in multiplication

int offset=0;

int dirPos=0;

while(dirPos!=val){

offset+=Integer.bitCount(dir[dirPos++]);

}

slot = (1<<(slot))-1; //turn slot into mask for N right bits

offset += 4+Integer.bitCount(dir[dirPos] & slot);

//turn into negative value if bit is not set, do not use conditions

return -offset + isSet*offset;

}

/** converts hash slot into actual offset in dir array, using bitmap */

protected static final int dirOffsetFromSlot(long[] dir, int slot) {

if(CC.ASSERT && slot>127)

throw new DBException.DataCorruption("slot too high");

int offset = 0;

long v = dir[0];

if(slot>63){

offset+=Long.bitCount(v);

v = dir[1];

}

slot &= 63;

long mask = ((1L)<<(slot&63))-1;

offset += 2+Long.bitCount(v & mask);

int v2 = (int) ((v>>>(slot))&1);

v2<<=1;

//turn into negative value if bit is not set, do not use conditions

return -offset + v2*offset;

}

protected static final Object dirPut(Object dir, int slot, long newRecid){

if(dir instanceof int[]) {

int[] dir_ = (int[]) dir;

int offset = dirOffsetFromSlot(dir_, slot);

//does new recid fit into integer?

if (newRecid <= Integer.MAX_VALUE) {

//make copy and expand it if necessary

if (offset < 0) {

offset = -offset;

dir_ = Arrays.copyOf(dir_, dir_.length + 1);

//make space for new value

System.arraycopy(dir_, offset, dir_, offset + 1, dir_.length - 1 - offset);

//and update bitmap

//TODO assert slot bit was not set

int bytePos = slot / 32;

int bitPos = slot % 32;

dir_[bytePos] = (dir_[bytePos] | (1 << bitPos));

} else {

//TODO assert slot bit was set

dir_ = dir_.clone();

}

//and insert value itself

dir_[offset] = (int) newRecid;

return dir_;

} else {

//new recid does not fit into long, so upgrade to long[] and continue

long[] dir2 = new long[dir_.length-2];

//bitmaps

dir2[0] = ((long)dir_[0]<<32) | dir_[1] & 0xFFFFFFFFL;

dir2[1] = ((long)dir_[2]<<32) | dir_[3] & 0xFFFFFFFFL;

for(int i=4;i<dir_.length;i++) {

dir2[i-2] = dir_[i];

}

dir = dir2;

}

}

//do long stuff

long[] dir_ = (long[]) dir;

int offset = dirOffsetFromSlot(dir_, slot);

//make copy and expand it if necessary

if (offset < 0) {

offset = -offset;

dir_ = Arrays.copyOf(dir_, dir_.length + 1);

//make space for new value

System.arraycopy(dir_, offset, dir_, offset + 1, dir_.length - 1 - offset);

//and update bitmap

//TODO assert slot bit was not set

int bytePos = slot / 64;

int bitPos = slot % 64;

dir_[bytePos] = (dir_[bytePos] | (1L << bitPos));

} else {

//TODO assert slot bit was set

dir_ = dir_.clone();

}

//and insert value itself

dir_[offset] = newRecid;

return dir_;

}

protected static final Object dirRemove(Object dir, final int slot){

int offset = dirOffsetFromSlot(dir, slot);

if(CC.ASSERT && offset<=0){

throw new DBException.DataCorruption("offset too low");

}

if(dir instanceof int[]) {

int[] dir_ = (int[]) dir;

//shrink and copy data

int[] dir2 = new int[dir_.length - 1];

System.arraycopy(dir_, 0, dir2, 0, offset);

System.arraycopy(dir_, offset + 1, dir2, offset, dir2.length - offset);

//unset bitmap bit

//TODO assert slot bit was set

int bytePos = slot / 32;

int bitPos = slot % 32;

dir2[bytePos] = (dir2[bytePos] & ~(1 << bitPos));

return dir2;

}else{

long[] dir_ = (long[]) dir;

//shrink and copy data

long[] dir2 = new long[dir_.length - 1];

System.arraycopy(dir_, 0, dir2, 0, offset);

System.arraycopy(dir_, offset + 1, dir2, offset, dir2.length - offset);

//unset bitmap bit

//TODO assert slot bit was set

int bytePos = slot / 64;

int bitPos = slot % 64;

dir2[bytePos] = (dir2[bytePos] & ~(1L << bitPos));

return dir2;

}

}

@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");

V ret;

final int h = hash(key);

final int segment = h >>>28;

consistencyLock.lock();

try {

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

try {

ret = putInner(key, value, h, segment);

} finally {

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

}

}finally {

consistencyLock.unlock();

}

if(expireSingleThreadFlag)

expirePurge();

return ret;

}

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

long dirRecid = segmentRecids[segment];

Engine engine = engines[segment];

int level = 3;

while(true){

Object dir = engine.get(dirRecid, DIR_SERIALIZER);

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

if(CC.ASSERT && slot>127)

throw new DBException.DataCorruption("slot too high");

if(dir == null ){

//create new dir

dir = new int[4];

}

final int dirOffset = dirOffsetFromSlot(dir,slot);

int counter = 0;

long recid = dirOffset<0 ? 0 : dirGet(dir, dirOffset);

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(keySerializer.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);

if(CC.ASSERT && ln.next==recid)

throw new DBException.DataCorruption("cyclic reference in linked list");

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));

if(CC.ASSERT && ln!=null && ln.next==recid)

throw new DBException.DataCorruption("cyclic reference in linked list");

counter++;

if(CC.ASSERT && counter>1024*1024)

throw new DBException.DataCorruption("linked list too large");

}

//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){

Object nextDir = new int[4];

{

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);

if(CC.ASSERT && newRecid==node.next)

throw new DBException.DataCorruption("cyclic reference in linked list");

//add newly inserted record

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

nextDir = dirPut(nextDir,pos,( newRecid<<1) | 1);

if(expireFlag)

expireLinkAdd(segment,expireNodeRecid,newRecid,h);

}

//redistribute linked bucket into new dir

long nodeRecid = dirOffset<0?0: dirGet(dir, dirOffset)>>>1;

while(nodeRecid!=0){

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

final long nextRecid = n.next;

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

final long recid2 = dirGetSlot(nextDir,pos);

n = new LinkedNode<K, V>(recid2>>>1, n.expireLinkNodeRecid, n.key, n.value);

nextDir = dirPut(nextDir,pos,(nodeRecid<<1) | 1);

engine.update(nodeRecid, n, LN_SERIALIZER);

if(CC.ASSERT && nodeRecid==n.next)

throw new DBException.DataCorruption("cyclic reference in linked list");

nodeRecid = nextRecid;

}

//insert nextDir and update parent dir

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

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

dir = dirPut(dir, parentPos, (nextDirRecid<<1) | 0);