implements ConcurrentNavigableMap<K,V>, Bind.MapWithModificationListener<K,V>, Closeable {

@SuppressWarnings("rawtypes")

public static final Comparator COMPARABLE_COMPARATOR = new Comparator<Comparable>() {

@Override

final public int compare(final Comparable o1, final Comparable o2) {

return o1.compareTo(o2);

}

};

protected static final Object EMPTY = new Object();

protected static final int B_TREE_NODE_LEAF_LR = 180;

protected static final int B_TREE_NODE_LEAF_L = 181;

protected static final int B_TREE_NODE_LEAF_R = 182;

protected static final int B_TREE_NODE_LEAF_C = 183;

protected static final int B_TREE_NODE_DIR_LR = 184;

protected static final int B_TREE_NODE_DIR_L = 185;

protected static final int B_TREE_NODE_DIR_R = 186;

protected static final int B_TREE_NODE_DIR_C = 187;

/** recid under which reference to rootRecid is stored */

protected final long rootRecidRef;

/** Serializer used to convert keys from/into binary form. */

protected final BTreeKeySerializer keySerializer;

/** Serializer used to convert keys from/into binary form*/

protected final Serializer<V> valueSerializer;

/** keys are sorted by this*/

protected final Comparator comparator;

/** holds node level locks*/

protected final LongConcurrentHashMap<Thread> nodeLocks = new LongConcurrentHashMap<Thread>();

/** maximal node size allowed in this BTree*/

protected final int maxNodeSize;

/** DB Engine in which entries are persisted */

protected final Engine engine;

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

protected final boolean hasValues;

/** store values as part of BTree nodes */

protected final boolean valsOutsideNodes;

protected final List<Long> leftEdges;

private final KeySet keySet;

private final EntrySet entrySet = new EntrySet(this);

private final Values values = new Values(this);

private final ConcurrentNavigableMap<K,V> descendingMap = new DescendingMap(this, null,true, null, false);

protected final Atomic.Long counter;

protected final int numberOfNodeMetas;

/** hack used for DB Catalog*/

protected static SortedMap<String, Object> preinitCatalog(DB db) {

Long rootRef = db.getEngine().get(Engine.CATALOG_RECID, Serializer.LONG);

if(rootRef==null){

if(db.getEngine().isReadOnly())

return Collections.unmodifiableSortedMap(new TreeMap<String, Object>());

NodeSerializer rootSerializer = new NodeSerializer(false,BTreeKeySerializer.STRING,

db.getDefaultSerializer(),COMPARABLE_COMPARATOR, 0);

BNode root = new LeafNode(new Object[]{null, null}, new Object[]{}, 0);

rootRef = db.getEngine().put(root, rootSerializer);

db.getEngine().update(Engine.CATALOG_RECID,rootRef, Serializer.LONG);

db.getEngine().commit();

}

return new BTreeMap<String, Object>(db.engine,Engine.CATALOG_RECID,32,false,0,

BTreeKeySerializer.STRING,

db.getDefaultSerializer(),

COMPARABLE_COMPARATOR,0, false);

}

/** if <code>valsOutsideNodes</code> is true, this class is used instead of values.

protected static final class ValRef{

/** reference to actual value */

final long recid;

public ValRef(long recid) {

this.recid = recid;

}

@Override

public boolean equals(Object obj) {

throw new IllegalAccessError();

}

@Override

public int hashCode() {

throw new IllegalAccessError();

}

@Override

public String toString() {

return "BTreeMap-ValRer["+recid+"]";

}

}

/** common interface for BTree node */

public interface BNode{

boolean isLeaf();

Object[] keys();

Object[] vals();

Object highKey();

long[] child();

long next();

}

public final static class DirNode implements BNode{

final Object[] keys;

final long[] child;

DirNode(Object[] keys, long[] child) {

this.keys = keys;

this.child = child;

}

DirNode(Object[] keys, List<Long> child) {

this.keys = keys;

this.child = new long[child.size()];

for(int i=0;i<child.size();i++){

this.child[i] = child.get(i);

}

}

@Override public boolean isLeaf() { return false;}

@Override public Object[] keys() { return keys;}

@Override public Object[] vals() { return null;}

@Override public Object highKey() {return keys[keys.length-1];}

@Override public long[] child() { return child;}

@Override public long next() {return child[child.length-1];}

@Override public String toString(){

return "Dir(K"+Arrays.toString(keys)+", C"+Arrays.toString(child)+")";

}

}

public final static class LeafNode implements BNode{

final Object[] keys;

final Object[] vals;

final long next;

LeafNode(Object[] keys, Object[] vals, long next) {

this.keys = keys;

this.vals = vals;

this.next = next;

assert(vals==null||keys.length == vals.length+2);

}

@Override public boolean isLeaf() { return true;}

@Override public Object[] keys() { return keys;}

@Override public Object[] vals() { return vals;}

@Override public Object highKey() {return keys[keys.length-1];}

@Override public long[] child() { return null;}

@Override public long next() {return next;}

@Override public String toString(){

return "Leaf(K"+Arrays.toString(keys)+", V"+Arrays.toString(vals)+", L="+next+")";

}

}

protected final Serializer<BNode> nodeSerializer;

protected static final class NodeSerializer<A,B> implements Serializer<BNode>{

protected final boolean hasValues;

protected final boolean valsOutsideNodes;

protected final BTreeKeySerializer keySerializer;

protected final Serializer<Object> valueSerializer;

protected final Comparator comparator;

protected final int numberOfNodeMetas;

public NodeSerializer(boolean valsOutsideNodes, BTreeKeySerializer keySerializer, Serializer valueSerializer, Comparator comparator, int numberOfNodeMetas) {

assert(keySerializer!=null);

assert(comparator!=null);

this.hasValues = valueSerializer!=null;

this.valsOutsideNodes = valsOutsideNodes;

this.keySerializer = keySerializer;

this.valueSerializer = valueSerializer;

this.comparator = comparator;

this.numberOfNodeMetas = numberOfNodeMetas;

}

@Override

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

final boolean isLeaf = value.isLeaf();

//first byte encodes if is leaf (first bite) and length (last seven bites)

assert(value.keys().length<=255);

assert(!(!isLeaf && value.child().length!= value.keys().length));

assert(!(isLeaf && hasValues && value.vals().length!= value.keys().length-2));

assert(!(!isLeaf && value.highKey()!=null && value.child()[value.child().length-1]==0));

//check node integrity in paranoid mode

if(CC.PARANOID){

int len = value.keys().length;

for(int i=value.keys()[0]==null?2:1;

i<(value.keys()[len-1]==null?len-1:len);

i++){

int comp = comparator.compare(value.keys()[i-1], value.keys()[i]);

int limit = i==len-1 ? 1:0 ;

if(comp>=limit){

throw new AssertionError("BTreeNode format error, wrong key order at #"+i+"\n"+value);

}

}

}

final boolean left = value.keys()[0] == null;

final boolean right = value.keys()[value.keys().length-1] == null;

final int header = makeHeader(isLeaf, left, right);

out.write(header);

out.write(value.keys().length);

//write node metas, right now this is ignored, but in future it could be used for counted btrees or aggregations

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

DataOutput2.packLong(out,0);

}

//longs go first, so it is possible to reconstruct tree without serializer

if(isLeaf){

DataOutput2.packLong(out, ((LeafNode) value).next);

}else{

for(long child : ((DirNode)value).child)

DataOutput2.packLong(out, child);

}

keySerializer.serialize(out,left?1:0,

right?value.keys().length-1:value.keys().length,

value.keys());

if(isLeaf){

if(hasValues){

for(Object val:value.vals()){

assert(val!=null);

if(valsOutsideNodes){

long recid = ((ValRef)val).recid;

DataOutput2.packLong(out, recid);

}else{

valueSerializer.serialize(out, val);

}

}

}else{

//write bits if values are null

boolean[] bools = new boolean[value.vals().length];

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

bools[i] = value.vals()[i]!=null;

}

//pack

byte[] bb = SerializerBase.booleanToByteArray(bools);

out.write(bb);

}

}

}

private int makeHeader(final boolean isLeaf, final boolean left, final boolean right) {

if(isLeaf){

if(right){

if(left)

return B_TREE_NODE_LEAF_LR;

else

return B_TREE_NODE_LEAF_R;

}else{

if(left)

return B_TREE_NODE_LEAF_L;

else

return B_TREE_NODE_LEAF_C;

}

}else{

if(right){

if(left)

return B_TREE_NODE_DIR_LR;

else

return B_TREE_NODE_DIR_R;

}else{

if(left)

return B_TREE_NODE_DIR_L;

else

return B_TREE_NODE_DIR_C;

}

}

}

@Override

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

final int header = in.readUnsignedByte();

final int size = in.readUnsignedByte();

//read node metas, right now this is ignored, but in future it could be used for counted btrees or aggregations

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

DataInput2.unpackLong(in);

}

//first bite indicates leaf

final boolean isLeaf =

header == B_TREE_NODE_LEAF_C || header == B_TREE_NODE_LEAF_L ||

header == B_TREE_NODE_LEAF_LR || header == B_TREE_NODE_LEAF_R;

final int start =

(header==B_TREE_NODE_LEAF_L || header == B_TREE_NODE_LEAF_LR || header==B_TREE_NODE_DIR_L || header == B_TREE_NODE_DIR_LR) ?

1:0;

final int end =

(header==B_TREE_NODE_LEAF_R || header == B_TREE_NODE_LEAF_LR || header==B_TREE_NODE_DIR_R || header == B_TREE_NODE_DIR_LR) ?

size-1:size;

if(isLeaf){

return deserializeLeaf(in, size, start, end);

}else{

return deserializeDir(in, size, start, end);

}

}

private BNode deserializeDir(final DataInput in, final int size, final int start, final int end) throws IOException {

final long[] child = new long[size];

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

child[i] = DataInput2.unpackLong(in);

final Object[] keys = keySerializer.deserialize(in, start,end,size);

assert(keys.length==size);

return new DirNode(keys, child);

}

private BNode deserializeLeaf(final DataInput in, final int size, final int start, final int end) throws IOException {

final long next = DataInput2.unpackLong(in);

final Object[] keys = keySerializer.deserialize(in, start,end,size);

assert(keys.length==size);

Object[] vals = new Object[size-2];

if(hasValues){

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

if(valsOutsideNodes){

long recid = DataInput2.unpackLong(in);

vals[i] = recid==0? null: new ValRef(recid);

}else{

vals[i] = valueSerializer.deserialize(in, -1);

}

}

}else{

//restore values which were deleted

boolean[] bools = SerializerBase.readBooleanArray(vals.length, in);

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

if(bools[i])

vals[i]=EMPTY;

}

}

return new LeafNode(keys, vals, next);

}

@Override

public int fixedSize() {

return -1;

}

}

/** Constructor used to create new BTreeMap.

public BTreeMap(Engine engine, long rootRecidRef,int maxNodeSize, boolean valsOutsideNodes, long counterRecid,

BTreeKeySerializer<K> keySerializer, Serializer<V> valueSerializer, Comparator<K> comparator,

int numberOfNodeMetas, boolean disableLocks) {

if(maxNodeSize%2!=0) throw new IllegalArgumentException("maxNodeSize must be dividable by 2");

if(maxNodeSize<6) throw new IllegalArgumentException("maxNodeSize too low");

if(maxNodeSize>126) throw new IllegalArgumentException("maxNodeSize too high");

if(rootRecidRef<=0||counterRecid<0 || numberOfNodeMetas<0) throw new IllegalArgumentException();

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

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

SerializerBase.assertSerializable(keySerializer);

SerializerBase.assertSerializable(valueSerializer);

SerializerBase.assertSerializable(comparator);

this.rootRecidRef = rootRecidRef;

this.hasValues = valueSerializer!=null;

this.valsOutsideNodes = valsOutsideNodes;

this.engine = engine;

this.maxNodeSize = maxNodeSize;

this.comparator = comparator;

this.numberOfNodeMetas = numberOfNodeMetas;

{

Comparator requiredComparator = keySerializer.getComparator();

if(requiredComparator!=null && !requiredComparator.equals(comparator))

throw new IllegalArgumentException("KeySerializers requires its own comparator");

}

this.keySerializer = keySerializer;

this.valueSerializer = valueSerializer;

this.nodeSerializer = new NodeSerializer(valsOutsideNodes,keySerializer,valueSerializer,comparator,numberOfNodeMetas);

this.keySet = new KeySet(this, hasValues);

if(counterRecid!=0){

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

Bind.size(this,counter);

}else{

this.counter = null;

}

//load left edge refs

ArrayList leftEdges2 = new ArrayList<Long>();

long r = engine.get(rootRecidRef,Serializer.LONG);

for(;;){

BNode n= engine.get(r,nodeSerializer);

leftEdges2.add(r);

if(n.isLeaf()) break;

r = n.child()[0];

}

Collections.reverse(leftEdges2);

leftEdges = new CopyOnWriteArrayList<Long>(leftEdges2);

}

/** creates empty root node and returns recid of its reference*/

static protected long createRootRef(Engine engine, BTreeKeySerializer keySer, Serializer valueSer, Comparator comparator, int numberOfNodeMetas){

final LeafNode emptyRoot = new LeafNode(new Object[]{null, null}, new Object[]{}, 0);

//empty root is serializer simpler way, so we can use dummy values

long rootRecidVal = engine.put(emptyRoot, new NodeSerializer(false,keySer, valueSer, comparator, numberOfNodeMetas));

return engine.put(rootRecidVal,Serializer.LONG);

}

/**

protected final int findChildren(final Object key, final Object[] keys) {

int left = 0;

if(keys[0] == null) left++;

int right = keys[keys.length-1] == null ? keys.length-1 : keys.length;

int middle;

// binary search

while (true) {

middle = (left + right) / 2;

if(keys[middle]==null) return middle; //null is positive infinitive

if (comparator.compare(keys[middle], key) < 0) {

left = middle + 1;

} else {

right = middle;

}

if (left >= right) {

return right;

}

}

}

@Override

public V get(Object key){

return (V) get(key, true);

}

protected Object get(Object key, boolean expandValue) {

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

K v = (K) key;

long current = engine.get(rootRecidRef, Serializer.LONG); //get root

BNode A = engine.get(current, nodeSerializer);

//dive until leaf

while(!A.isLeaf()){

current = nextDir((DirNode) A, v);

A = engine.get(current, nodeSerializer);

}

//now at leaf level

LeafNode leaf = (LeafNode) A;

int pos = findChildren(v, leaf.keys);

while(pos == leaf.keys.length){

//follow next link on leaf until necessary

leaf = (LeafNode) engine.get(leaf.next, nodeSerializer);

pos = findChildren(v, leaf.keys);

}

if(pos==leaf.keys.length-1){

return null; //last key is always deleted

}

//finish search

if(leaf.keys[pos]!=null && 0==comparator.compare(v,leaf.keys[pos])){

Object ret = leaf.vals[pos-1];

return expandValue ? valExpand(ret) : ret;

}else

return null;

}

protected V valExpand(Object ret) {

if(valsOutsideNodes && ret!=null) {

long recid = ((ValRef)ret).recid;

ret = engine.get(recid, valueSerializer);

}

return (V) ret;

}

protected final long nextDir(DirNode d, Object key) {

int pos = findChildren(key, d.keys) - 1;

if(pos<0) pos = 0;

return d.child[pos];

}

@Override

public V put(K key, V value){

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

return put2(key,value, false);

}

protected V put2(final K key, final V value2, final boolean putOnlyIfAbsent){

K v = key;

if(v == null) throw new IllegalArgumentException("null key");

if(value2 == null) throw new IllegalArgumentException("null value");

V value = value2;

if(valsOutsideNodes){

long recid = engine.put(value2, valueSerializer);

value = (V) new ValRef(recid);

}

int stackPos = -1;

long[] stackVals = new long[4];

final long rootRecid = engine.get(rootRecidRef, Serializer.LONG);

long current = rootRecid;

BNode A = engine.get(current, nodeSerializer);

while(!A.isLeaf()){

long t = current;

current = nextDir((DirNode) A, v);

assert(current>0) : A;

if(current == A.child()[A.child().length-1]){

//is link, do nothing

}else{

//stack push t

stackPos++;

if(stackVals.length == stackPos) //grow if needed

stackVals = Arrays.copyOf(stackVals, stackVals.length*2);

stackVals[stackPos] = t;

}

A = engine.get(current, nodeSerializer);

}

int level = 1;

long p=0;

try{

while(true){

boolean found;

do{

lock(nodeLocks, current);

found = true;

A = engine.get(current, nodeSerializer);

int pos = findChildren(v, A.keys());

//check if keys is already in tree

if(pos<A.keys().length-1 && v!=null && A.keys()[pos]!=null &&

0==comparator.compare(v,A.keys()[pos])){

//yes key is already in tree

Object oldVal = A.vals()[pos-1];

if(putOnlyIfAbsent){

//is not absent, so quit

unlock(nodeLocks, current);

if(CC.PARANOID) assertNoLocks(nodeLocks);

return valExpand(oldVal);

}

//insert new

Object[] vals = Arrays.copyOf(A.vals(), A.vals().length);

vals[pos-1] = value;

A = new LeafNode(Arrays.copyOf(A.keys(), A.keys().length), vals, ((LeafNode)A).next);

assert(nodeLocks.get(current)==Thread.currentThread());

engine.update(current, A, nodeSerializer);

//already in here

V ret = valExpand(oldVal);

notify(key,ret, value2);

unlock(nodeLocks, current);

if(CC.PARANOID) assertNoLocks(nodeLocks);

return ret;

}

//if v > highvalue(a)

if(A.highKey() != null && comparator.compare(v, A.highKey())>0){

//follow link until necessary

unlock(nodeLocks, current);

found = false;

int pos2 = findChildren(v, A.keys());

while(A!=null && pos2 == A.keys().length){

//TODO lock?

long next = A.next();

if(next==0) break;

current = next;

A = engine.get(current, nodeSerializer);

pos2 = findChildren(v, A.keys());

}

}

}while(!found);

// can be new item inserted into A without splitting it?

if(A.keys().length - (A.isLeaf()?2:1)<maxNodeSize){

int pos = findChildren(v, A.keys());

Object[] keys = arrayPut(A.keys(), pos, v);

if(A.isLeaf()){

Object[] vals = arrayPut(A.vals(), pos-1, value);

LeafNode n = new LeafNode(keys, vals, ((LeafNode)A).next);

assert(nodeLocks.get(current)==Thread.currentThread());

engine.update(current, n, nodeSerializer);

}else{

assert(p!=0);

long[] child = arrayLongPut(A.child(), pos, p);

DirNode d = new DirNode(keys, child);

assert(nodeLocks.get(current)==Thread.currentThread());

engine.update(current, d, nodeSerializer);

}

notify(key, null, value2);

unlock(nodeLocks, current);

if(CC.PARANOID) assertNoLocks(nodeLocks);

return null;

}else{

//node is not safe, it requires splitting

final int pos = findChildren(v, A.keys());

final Object[] keys = arrayPut(A.keys(), pos, v);

final Object[] vals = (A.isLeaf())? arrayPut(A.vals(), pos-1, value) : null;

final long[] child = A.isLeaf()? null : arrayLongPut(A.child(), pos, p);

final int splitPos = keys.length/2;

BNode B;

if(A.isLeaf()){

Object[] vals2 = Arrays.copyOfRange(vals, splitPos, vals.length);

B = new LeafNode(

Arrays.copyOfRange(keys, splitPos, keys.length),

vals2,

((LeafNode)A).next);

}else{

B = new DirNode(Arrays.copyOfRange(keys, splitPos, keys.length),

Arrays.copyOfRange(child, splitPos, keys.length));

}

long q = engine.put(B, nodeSerializer);

if(A.isLeaf()){ // splitPos+1 is there so A gets new high value (key)

Object[] keys2 = Arrays.copyOf(keys, splitPos+2);

keys2[keys2.length-1] = keys2[keys2.length-2];

Object[] vals2 = Arrays.copyOf(vals, splitPos);

//TODO check high/low keys overlap

A = new LeafNode(keys2, vals2, q);

}else{

long[] child2 = Arrays.copyOf(child, splitPos+1);

child2[splitPos] = q;

A = new DirNode(Arrays.copyOf(keys, splitPos+1), child2);

}

assert(nodeLocks.get(current)==Thread.currentThread());

engine.update(current, A, nodeSerializer);

if((current != rootRecid)){ //is not root

unlock(nodeLocks, current);

p = q;

v = (K) A.highKey();

level = level+1;

if(stackPos!=-1){ //if stack is not empty

current = stackVals[stackPos--];

}else{

//current := the left most node at level

current = leftEdges.get(level-1);

}

assert(current>0);

}else{

BNode R = new DirNode(

new Object[]{A.keys()[0], A.highKey(), B.isLeaf()?null:B.highKey()},

new long[]{current,q, 0});

lock(nodeLocks, rootRecidRef);

unlock(nodeLocks, current);

long newRootRecid = engine.put(R, nodeSerializer);

assert(nodeLocks.get(rootRecidRef)==Thread.currentThread());

engine.update(rootRecidRef, newRootRecid, Serializer.LONG);

//add newRootRecid into leftEdges

leftEdges.add(newRootRecid);

notify(key, null, value2);

unlock(nodeLocks, rootRecidRef);

if(CC.PARANOID) assertNoLocks(nodeLocks);

return null;

}

}

}

}catch(RuntimeException e){

unlockAll(nodeLocks);

throw e;

}catch(Exception e){

unlockAll(nodeLocks);

throw new RuntimeException(e);

}

}

protected static class BTreeIterator{

final BTreeMap m;

LeafNode currentLeaf;

Object lastReturnedKey;

int currentPos;

final Object hi;

final boolean hiInclusive;

/** unbounded iterator*/

BTreeIterator(BTreeMap m){

this.m = m;

hi=null;

hiInclusive=false;

pointToStart();

}

/** bounder iterator, args may be null for partially bounded*/

BTreeIterator(BTreeMap m, Object lo, boolean loInclusive, Object hi, boolean hiInclusive){

this.m = m;

if(lo==null){

pointToStart();

}else{

Fun.Tuple2<Integer, LeafNode> l = m.findLargerNode(lo, loInclusive);

currentPos = l!=null? l.a : -1;

currentLeaf = l!=null ? l.b : null;

}

this.hi = hi;

this.hiInclusive = hiInclusive;

if(hi!=null && currentLeaf!=null){

//check in bounds

Object key = currentLeaf.keys[currentPos];

int c = m.comparator.compare(key, hi);

if (c > 0 || (c == 0 && !hiInclusive)){

//out of high bound

currentLeaf=null;

currentPos=-1;

}

}

}

private void pointToStart() {

//find left-most leaf

final long rootRecid = m.engine.get(m.rootRecidRef, Serializer.LONG);

BNode node = (BNode) m.engine.get(rootRecid, m.nodeSerializer);

while(!node.isLeaf()){

node = (BNode) m.engine.get(node.child()[0], m.nodeSerializer);

}

currentLeaf = (LeafNode) node;

currentPos = 1;

while(currentLeaf.keys.length==2){

//follow link until leaf is not empty

if(currentLeaf.next == 0){

currentLeaf = null;

return;

}

currentLeaf = (LeafNode) m.engine.get(currentLeaf.next, m.nodeSerializer);

}

}

public boolean hasNext(){

return currentLeaf!=null;

}

public void remove(){

if(lastReturnedKey==null) throw new IllegalStateException();

m.remove(lastReturnedKey);

lastReturnedKey = null;

}

protected void advance(){

if(currentLeaf==null) return;

lastReturnedKey = currentLeaf.keys[currentPos];

currentPos++;

if(currentPos == currentLeaf.keys.length-1){

//move to next leaf

if(currentLeaf.next==0){

currentLeaf = null;

currentPos=-1;

return;

}

currentPos = 1;

currentLeaf = (LeafNode) m.engine.get(currentLeaf.next, m.nodeSerializer);

while(currentLeaf.keys.length==2){

if(currentLeaf.next ==0){

currentLeaf = null;

currentPos=-1;

return;

}

currentLeaf = (LeafNode) m.engine.get(currentLeaf.next, m.nodeSerializer);

}

}

if(hi!=null && currentLeaf!=null){

//check in bounds

Object key = currentLeaf.keys[currentPos];

int c = m.comparator.compare(key, hi);

if (c > 0 || (c == 0 && !hiInclusive)){

//out of high bound

currentLeaf=null;

currentPos=-1;

}

}

}

}

@Override

public V remove(Object key) {

return remove2(key, null);

}

private V remove2(final Object key, final Object value) {

long current = engine.get(rootRecidRef, Serializer.LONG);

BNode A = engine.get(current, nodeSerializer);

while(!A.isLeaf()){

current = nextDir((DirNode) A, key);

A = engine.get(current, nodeSerializer);

}

try{

while(true){

lock(nodeLocks, current);

A = engine.get(current, nodeSerializer);

int pos = findChildren(key, A.keys());

if(pos<A.keys().length&& key!=null && A.keys()[pos]!=null &&

0==comparator.compare(key,A.keys()[pos])){

//check for last node which was already deleted

if(pos == A.keys().length-1 && value == null){

unlock(nodeLocks, current);

return null;

}

//delete from node

Object oldVal = A.vals()[pos-1];

oldVal = valExpand(oldVal);

if(value!=null && !value.equals(oldVal)){

unlock(nodeLocks, current);

return null;

}

Object[] keys2 = new Object[A.keys().length-1];

System.arraycopy(A.keys(),0,keys2, 0, pos);

System.arraycopy(A.keys(), pos+1, keys2, pos, keys2.length-pos);

Object[] vals2 = new Object[A.vals().length-1];

System.arraycopy(A.vals(),0,vals2, 0, pos-1);