adressing core review, adding some comments for some of the cases

core/src/main/java/com/yahoo/oak/EntryOrderedSet.java

@@ -297,11 +297,13 @@ boolean isEntrySetValidAfterRebalance() {
         return true;
     }
 
+    //This method is called by only one thread on a release chunk.
     void releaseAllDeletedKeys() {
         KeyBuffer key = new KeyBuffer(config.keysMemoryManager.getEmptySlice());
         for (int i = 0; i < nextFreeIndex.get() - 1 && i < numOfEntries.get() ; i++) {
-            //the second condiiton is for case where multiple threads increment the nextfreeIndex
-            //which results in being bigger than the array size
+            //the second condition is for case where multiple threads increment the nextfreeIndex
+            //which results in being bigger than the array size, this happens when multiple threads try
+            // to increase the nextFreeIndex
             if (!config.valuesMemoryManager.isReferenceDeleted(getValueReference(i))) {
                 continue;
             }

core/src/main/java/com/yahoo/oak/InternalOakBasics.java

@@ -165,11 +165,14 @@ V replace(K key, V value, OakTransformer<V> valueDeserializeTransformer) {
         ThreadContext ctx = getThreadContext();
 
         for (int i = 0; i < MAX_RETRIES; i++) {
-            BasicChunk<K, V> chunk = findChunk(key, ctx); // find orderedChunk matching key
+            BasicChunk<K, V> chunk = findChunk(key, ctx); // find Chunk matching key
             try {
                 chunk.lookUp(ctx, key);
+                //the look up method might encounter a chunk which is released, while using Nova as a memory manager
+                //as a result we might access an already deleted key, thus the need to catch the exception 
             } catch (DeletedMemoryAccessException e) {
-                assert !(chunk instanceof  HashChunk);
+                assert !(chunk instanceof  HashChunk); 
+                //Hash deals with deleted keys in an earlier stage
                 continue;
             }
             if (!ctx.isValueValid()) {
@@ -338,7 +341,8 @@ public boolean hasNext() {
             return (state != null);
         }
 
-        protected abstract void initStateWithNotDeletedChunk(BasicChunk chunk);
+        // for more detail on this method see implementation 
+        protected abstract void initStateWithMinKey(BasicChunk<K, V> chunk);
 
         protected abstract void initAfterRebalance();
 
@@ -375,7 +379,7 @@ public void remove() {
          * The first long is the key's reference, the integer is the value's version and the second long is
          * the value's reference. If {@code needsValue == false}, then the value of the map entry is {@code null}.
          */
-        void advance(boolean needsValue) throws DeletedMemoryAccessException {
+        void advance(boolean needsValue) {
             boolean validState = false;
 
             while (!validState) {
@@ -481,7 +485,7 @@ protected boolean advanceState() {
                     chunkIter = getChunkIter(chunk);
                 } catch (DeletedMemoryAccessException e) {
                     assert chunk.state.get() ==  BasicChunk.State.RELEASED;
-                    initStateWithNotDeletedChunk(chunk); 
+                    initStateWithMinKey(chunk); 
                     return true;
                 }
             }

core/src/main/java/com/yahoo/oak/InternalOakHash.java

@@ -268,12 +268,8 @@ boolean replace(K key, V oldValue, V newValue, OakTransformer<V> valueDeserializ
 
         for (int i = 0; i < MAX_RETRIES; i++) {
             // find chunk matching key, puts this key hash into ctx.operationKeyHash
-            BasicChunk<K, V> c = findChunk(key, ctx);
-            try {
-                c.lookUp(ctx, key);
-            } catch (DeletedMemoryAccessException e) {
-                continue;
-            }
+            HashChunk<K, V> c = findChunk(key, ctx);
+            c.lookUp(ctx, key);
             if (!ctx.isValueValid()) {
                 return false;
             }
@@ -410,12 +406,8 @@ boolean computeIfPresent(K key, Consumer<OakScopedWriteBuffer> computer) {
 
         for (int i = 0; i < MAX_RETRIES; i++) {
             // find chunk matching key, puts this key hash into ctx.operationKeyHash
-            BasicChunk<K, V> c = findChunk(key, ctx);
-            try {
-                c.lookUp(ctx, key);
-            } catch (DeletedMemoryAccessException e) {
-                continue;
-            }
+            HashChunk<K, V> c = findChunk(key, ctx);
+            c.lookUp(ctx, key);
             if (ctx.isValueValid()) {
                 ValueUtils.ValueResult res = config.valueOperator.compute(ctx.value, computer);
                 if (res == ValueUtils.ValueResult.TRUE) {
@@ -447,12 +439,8 @@ Result remove(K key, V oldValue, OakTransformer<V> transformer) {
 
         for (int i = 0; i < MAX_RETRIES; i++) {
             // find chunk matching key, puts this key hash into ctx.operationKeyHash
-            BasicChunk<K, V> c = findChunk(key, ctx);
-            try {
-                c.lookUp(ctx, key);
-            } catch (DeletedMemoryAccessException e) {
-                continue;
-            }
+            HashChunk<K, V> c = findChunk(key, ctx);
+            c.lookUp(ctx, key);
             if (!ctx.isKeyValid()) {
                 // There is no such key. If we did logical deletion and someone else did the physical deletion,
                 // then the old value is saved in v. Otherwise v is (correctly) null
@@ -580,7 +568,7 @@ protected void initAfterRebalance() {
         }
 
         @Override
-        protected void initStateWithNotDeletedChunk(BasicChunk c) { //nextKey is null here
+        protected void initStateWithMinKey(BasicChunk c) { //nextKey is null here
             initState();
         }
 

core/src/main/java/com/yahoo/oak/InternalOakMap.java

@@ -982,14 +982,16 @@ protected void initAfterRebalance() {
             initState(isDescending, lo, loInclusive, hi, hiInclusive);
         }
 
+        //The same as initState, but always works on a released chunk minKey which is never deleted.
         @Override
-        protected void initStateWithNotDeletedChunk(BasicChunk chunk) {
-            OrderedChunk oChunk = (OrderedChunk) chunk;
+        protected void initStateWithMinKey(BasicChunk <K, V> chunk) {
+            OrderedChunk <K, V> oChunk = (OrderedChunk <K, V>) chunk;
             K nextKey = null;
             try {
                 nextKey = KeyUtils.deSerializedKey(oChunk.minKey, getKeySerializer());
             } catch (DeletedMemoryAccessException e) {
-                return ; //should not happen
+                assert e == null; //since we are not deleting minKeys (should not get here!)
+                return ; 
             }
             if (isDescending) {
                 hiInclusive = true;
@@ -1138,7 +1140,10 @@ protected void initState(boolean isDescending, K lowerBound, boolean lowerInclus
                     setState(IteratorState.newInstance(nextOrderedChunk, nextChunkIter));
                     advanceState();
                 } catch (DeletedMemoryAccessException e) {
-                    continue;
+                    assert e == null; 
+                    //The exception here is needed as the ascendingIter/descendingIter signature throws the 
+                    // exception in case that the key is deleted, this could not happen here since we are dealing with
+                    // minKey which we never delete even when using memory manager that deleted keys
                 }
                 return;
             }
@@ -1147,11 +1152,11 @@ protected void initState(boolean isDescending, K lowerBound, boolean lowerInclus
 
         @Override
         protected BasicChunk<K, V> getNextChunk(BasicChunk<K, V> current) {
-            OrderedChunk<K, V> currenChunk = (OrderedChunk<K, V>) current;
+            OrderedChunk<K, V> currentChunk = (OrderedChunk<K, V>) current;
             if (!isDescending) {
-                return  currenChunk.next.getReference();
+                return  currentChunk.next.getReference();
             } else {
-                Map.Entry<Object, OrderedChunk<K, V>> entry = skiplist.lowerEntry(currenChunk.minKey);
+                Map.Entry<Object, OrderedChunk<K, V>> entry = skiplist.lowerEntry(currentChunk.minKey);
                 if (entry == null) {
                     return null;
                 } else {
@@ -1160,6 +1165,8 @@ protected BasicChunk<K, V> getNextChunk(BasicChunk<K, V> current) {
             }
         }
 
+        //This method throws DeletedMemoryAccessException when checking if some deleted key,
+        // is in the boundary of some released chunk.
         protected BasicChunk.BasicChunkIter getChunkIter(BasicChunk<K, V> current) throws DeletedMemoryAccessException {
             if (!isDescending) {
                 OakScopedReadBuffer upperBoundKeyForChunk = getNextChunkMinKey((OrderedChunk<K, V>) current);

core/src/main/java/com/yahoo/oak/OrderedChunk.java

@@ -26,7 +26,8 @@ class OrderedChunk<K, V> extends BasicChunk<K, V> {
     public static final int ORDERED_CHUNK_MAX_ITEMS_DEFAULT = 4096;
 
     /*-------------- Members --------------*/
-    KeyBuffer minKey;       // minimal key that can be put in this chunk
+    KeyBuffer minKey;       // minimal key that can be put in this chunk.
+    //These keys are never released even when using Nova memory manager
     AtomicMarkableReference<OrderedChunk<K, V>> next;
     private final EntryOrderedSet<K, V> entryOrderedSet;
 
@@ -740,11 +741,7 @@ class AscendingIter extends ChunkIter {
             OakScopedReadBuffer nextChunkMinKey) throws DeletedMemoryAccessException {
             next = entryOrderedSet.getHeadNextEntryIndex();
             next = advanceNextIndexNoBound(next, ctx);
-            try {
-                setIsEndBoundCheckNeeded(ctx, to, toInclusive, nextChunkMinKey);
-            } catch (DeletedMemoryAccessException e) {
-                throw new DeletedMemoryAccessException(); //TODO fix
-            }
+            setIsEndBoundCheckNeeded(ctx, to, toInclusive, nextChunkMinKey);
         }
 
         AscendingIter(ThreadContext ctx, K from, boolean fromInclusive, K to, boolean toInclusive,