DeletedMemoryAccessException is Exception

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

@@ -66,7 +66,10 @@ protected void updateBasicChild(BasicChunk<K, V> child) {
 
     abstract boolean readValueFromEntryIndex(ValueBuffer value, int ei);
 
-    abstract void lookUp(ThreadContext ctx, K key);
+    /* the lookUp method throws this exception for OrderedChunks only while trying to access a released chunk.
+     * the lookUp method is not allowed to throw the exception in the case of HashChunk.
+     */
+    abstract void lookUp(ThreadContext ctx, K key) throws DeletedMemoryAccessException;
 
     /*-------------- Publishing related methods and getters ---------------*/
     /**

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

@@ -6,7 +6,11 @@
 
 package com.yahoo.oak;
 
-public class DeletedMemoryAccessException extends RuntimeException {
+/**
+ * This Exception is thrown when using the Nova MemoryManager, such exception indicates that we are trying
+ * to access some off heap memory location that was deleted concurrently to the access.
+ */
+public class DeletedMemoryAccessException extends Exception {
     public DeletedMemoryAccessException() {
         super();
     }

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

@@ -182,9 +182,9 @@ int getNextNonZeroIndex(int currentIndex) {
         int nextIndex = currentIndex;
         while (!isNotZero && isIndexInBound(++nextIndex)) {
 
-            isNotZero = getEntryFieldLong(nextIndex, KEY_REF_OFFSET) != 0 ||
-                        getEntryFieldLong(nextIndex, VALUE_REF_OFFSET) != 0 ||
-                        getEntryFieldLong(nextIndex, OPT_FIELD_OFFSET) != 0;
+            isNotZero = array.getEntryFieldLong(nextIndex, KEY_REF_OFFSET) != 0 ||
+                        array.getEntryFieldLong(nextIndex, VALUE_REF_OFFSET) != 0 ||
+                        array.getEntryFieldLong(nextIndex, OPT_FIELD_OFFSET) != 0;
         }
         if (isIndexInBound(nextIndex)) {
             return nextIndex;

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

@@ -299,7 +299,9 @@ boolean isEntrySetValidAfterRebalance() {
 
     void releaseAllDeletedKeys() {
         KeyBuffer key = new KeyBuffer(config.keysMemoryManager.getEmptySlice());
-        for (int i = 0; i < numOfEntries.get() ; i++) {
+        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
             if (!config.valuesMemoryManager.isReferenceDeleted(getValueReference(i))) {
                 continue;
             }

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

@@ -166,7 +166,12 @@ V replace(K key, V value, OakTransformer<V> valueDeserializeTransformer) {
 
         for (int i = 0; i < MAX_RETRIES; i++) {
             BasicChunk<K, V> chunk = findChunk(key, ctx); // find orderedChunk matching key
-            chunk.lookUp(ctx, key);
+            try {
+                chunk.lookUp(ctx, key);
+            } catch (DeletedMemoryAccessException e) {
+                assert !(chunk instanceof  HashChunk);
+                continue;
+            }
             if (!ctx.isValueValid()) {
                 return null;
             }
@@ -184,29 +189,7 @@ V replace(K key, V value, OakTransformer<V> valueDeserializeTransformer) {
         throw new RuntimeException("replace failed: reached retry limit (1024).");
     }
 
-    boolean replace(K key, V oldValue, V newValue, OakTransformer<V> valueDeserializeTransformer) {
-        ThreadContext ctx = getThreadContext();
-
-        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); // find orderedChunk matching key
-            c.lookUp(ctx, key);
-            if (!ctx.isValueValid()) {
-                return false;
-            }
-
-            ValueUtils.ValueResult res = getValueOperator().compareExchange(c, ctx, oldValue, newValue,
-                    valueDeserializeTransformer, getValueSerializer());
-            if (res == ValueUtils.ValueResult.RETRY) {
-                // it might be that this chunk is proceeding with rebalance -> help
-                helpRebalanceIfInProgress(c);
-                continue;
-            }
-            return res == ValueUtils.ValueResult.TRUE;
-        }
-
-        throw new RuntimeException("replace failed: reached retry limit (1024).");
-    }
+    abstract boolean replace(K key, V oldValue, V newValue, OakTransformer<V> valueDeserializeTransformer);
 
     abstract boolean putIfAbsentComputeIfPresent(K key, V value, Consumer<OakScopedWriteBuffer> computer);
 
@@ -355,6 +338,8 @@ public boolean hasNext() {
             return (state != null);
         }
 
+        protected abstract void initStateWithNotDeletedChunk(BasicChunk chunk);
+
         protected abstract void initAfterRebalance();
 
         // the actual next()
@@ -390,7 +375,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) {
+        void advance(boolean needsValue) throws DeletedMemoryAccessException {
             boolean validState = false;
 
             while (!validState) {
@@ -471,7 +456,8 @@ protected void invalidatePrevState() {
 
         protected abstract BasicChunk<K, V> getNextChunk(BasicChunk<K, V> current);
 
-        protected abstract BasicChunk.BasicChunkIter getChunkIter(BasicChunk<K, V> current);
+        protected abstract BasicChunk.BasicChunkIter getChunkIter(BasicChunk<K, V> current)
+                throws DeletedMemoryAccessException;
 
         /**
          * advance state to the new position
@@ -483,21 +469,27 @@ protected boolean advanceState() {
             BasicChunk.BasicChunkIter chunkIter = getState().getChunkIter();
 
             updatePreviousState();
-
+            
             while (!chunkIter.hasNext()) { // skip empty chunks
                 chunk = getNextChunk(chunk);
                 if (chunk == null) {
                     //End of iteration
                     setState(null);
                     return false;
                 }
-                chunkIter = getChunkIter(chunk);
+                try {
+                    chunkIter = getChunkIter(chunk);
+                } catch (DeletedMemoryAccessException e) {
+                    assert chunk.state.get() ==  BasicChunk.State.RELEASED;
+                    initStateWithNotDeletedChunk(chunk); 
+                    return true;
+                }
             }
 
             int nextIndex = chunkIter.next(ctx);
             getState().set(chunk, chunkIter, nextIndex);
 
-            return true;
+            return true; 
         }
 
     }

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

@@ -269,7 +269,11 @@ 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);
-            c.lookUp(ctx, key);
+            try {
+                c.lookUp(ctx, key);
+            } catch (DeletedMemoryAccessException e) {
+                continue;
+            }
             if (!ctx.isValueValid()) {
                 return false;
             }
@@ -407,8 +411,11 @@ 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);
-            c.lookUp(ctx, key);
-
+            try {
+                c.lookUp(ctx, key);
+            } catch (DeletedMemoryAccessException e) {
+                continue;
+            }
             if (ctx.isValueValid()) {
                 ValueUtils.ValueResult res = config.valueOperator.compute(ctx.value, computer);
                 if (res == ValueUtils.ValueResult.TRUE) {
@@ -441,8 +448,11 @@ 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);
-            c.lookUp(ctx, key);
-
+            try {
+                c.lookUp(ctx, key);
+            } catch (DeletedMemoryAccessException e) {
+                continue;
+            }
             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
@@ -569,6 +579,10 @@ protected void initAfterRebalance() {
             initState();
         }
 
+        @Override
+        protected void initStateWithNotDeletedChunk(BasicChunk c) { //nextKey is null here
+            initState();
+        }
 
         /**
          * Advances next to higher entry.
@@ -799,8 +813,9 @@ class EntryTransformIterator<T> extends HashIter<T> {
         }
 
         public T next() {
+            ValueUtils.ValueResult res;
             advance(true);
-            ValueUtils.ValueResult res = ctx.value.s.preRead();
+            res = ctx.value.s.preRead();        
             if (res == ValueUtils.ValueResult.FALSE) {
                 return next();
             } else if (res == ValueUtils.ValueResult.RETRY) {
@@ -817,7 +832,11 @@ public T next() {
                     new AbstractMap.SimpleEntry<>(ctx.key, ctx.value);
 
             T transformation = transformer.apply(entry);
-            ctx.value.s.postRead();
+            try {
+                ctx.value.s.postRead();
+            } catch (DeletedMemoryAccessException e) {
+                return next();
+            }
             return transformation;
         }
     }