Make IterationContexts' hashLookupPos depend on segmentTier as well as QueryContexts'; add missed hashLookupPos step back in iteration, if hash lookup slot is removed, because entry is relocated to a chained tier; add missed tierIndex initialization in initSegmentTier_WithBaseAddr()

2021-09-13 01:26:12 +00:00 · 2015-10-07 16:43:37 +03:00
parent 03a761ebbf
commit 81535163b2
8 changed files with 96 additions and 52 deletions
@@ -25,10 +25,20 @@ import net.openhft.sg.Staged;
@Staged
 public abstract class HashLookupPos {

+    @StageRef HashLookupSearch hls;
+
    public long hashLookupPos = -1;

    public abstract boolean hashLookupPosInit();

+    public void initHashLookupPos() {
+        // validation + make hashLookupPos a dependant of segmentTier. This is needed, because
+        // after tier change should re-perform hashLookupSearch, starting from the searchStartPos
+        assert s.segmentTier >= 0;
+        s.innerReadLock.lock();
+        this.hashLookupPos = hls.searchStartPos;
+    }
+
    public void initHashLookupPos(long hashLookupPos) {
        this.hashLookupPos = hashLookupPos;
    }
@@ -22,7 +22,6 @@ import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable;
 import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
 import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages;
 import net.openhft.chronicle.hash.impl.stage.entry.HashLookupPos;
-import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
 import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
 import net.openhft.sg.StageRef;
 import net.openhft.sg.Staged;
@@ -73,7 +72,9 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
            while (true) {
                int currentTier = s.segmentTier;
                long currentTierBaseAddr = s.segmentBaseAddr;
-                size = forEachTierWhile(action, size, currentTier, currentTierBaseAddr);
+                long currentTierIndex = s.tierIndex;
+                size = forEachTierWhile(action, size,
+                        currentTier, currentTierBaseAddr, currentTierIndex);
                if (size < 0) // interrupted
                    return false;
                if (size == 0)
@@ -92,17 +93,24 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
    }

    private long forEachTierWhile(
-            Predicate<? super E> action, long size, int currentTier, long currentTierBaseAddr) {
+            Predicate<? super E> action, long size,
+            int currentTier, long currentTierBaseAddr, long tierIndex) {
        boolean interrupted = false;
        long startPos = 0L;
        CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup;
-        while (!hashLookup.empty(hashLookup.readEntry(s.segmentBaseAddr, startPos))) {
+        while (!hashLookup.empty(hashLookup.readEntry(currentTierBaseAddr, startPos))) {
            startPos = hashLookup.step(startPos);
        }
        hlp.initHashLookupPos(startPos);
+        long currentHashLookupPos;
        do {
-            hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos));
-            long entry = hashLookup.readEntry(s.segmentBaseAddr, hlp.hashLookupPos);
+            // Step from hlp.hashLookupPos, not currentHashLookupPos (with additional initialization
+            // of this local variable to startPos outside the loop), because if e.remove() is
+            // called in the `action`, hlp.hashLookupPos is stepped back in doRemove(), and
+            // currentHashLookupPos become invalid
+            currentHashLookupPos = hashLookup.step(hlp.hashLookupPos);
+            hlp.setHashLookupPos(currentHashLookupPos);
+            long entry = hashLookup.readEntry(currentTierBaseAddr, currentHashLookupPos);
            initHashLookupEntry(entry);
            if (!hashLookup.empty(entry)) {
                e.readExistingEntry(hashLookup.value(entry));
@@ -119,12 +127,20 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
                    } finally {
                        // if doReplaceValue() -> relocation() -> alloc() -> nextTier()
                        // was called, restore the tier we were iterating over
-                        if (s.segmentTier != currentTier)
-                            s.initSegmentTier_WithBaseAddr(currentTier, currentTierBaseAddr);
+                        if (s.segmentTier != currentTier) {
+                            s.initSegmentTier_WithBaseAddr(
+                                    currentTier, currentTierBaseAddr, tierIndex);
+                            // To cover shift deleted slot, at the next step forward.
+                            // hash lookup entry is relocated to the next chained tier, and the
+                            // slot in _current_ tier's hash lookup is shift deleted, see
+                            // relocation()
+                            currentHashLookupPos = hashLookup.stepBack(currentHashLookupPos);
+                            hlp.initHashLookupPos(currentHashLookupPos);
+                        }
                    }
                }
            }
-        } while (hlp.hashLookupPos != startPos);
+        } while (currentHashLookupPos != startPos);
        return interrupted ? ~size : size;
    }

@@ -45,8 +45,9 @@ public abstract class IterationSegmentStages extends SegmentStages {
            hls.initSearchKey(hh.h().hashLookup.key(it.hashLookupEntry));
    }

-    public void initSegmentTier_WithBaseAddr(int tier, long tierBaseAddr) {
+    public void initSegmentTier_WithBaseAddr(int tier, long tierBaseAddr, long tierIndex) {
        segmentTier = tier;
+        this.tierIndex = tierIndex;
        segmentBaseAddr = tierBaseAddr;
    }

@@ -1,39 +0,0 @@
-/*
- *      Copyright (C) 2015  higherfrequencytrading.com
- *
- *      This program is free software: you can redistribute it and/or modify
- *      it under the terms of the GNU Lesser General Public License as published by
- *      the Free Software Foundation, either version 3 of the License.
- *
- *      This program is distributed in the hope that it will be useful,
- *      but WITHOUT ANY WARRANTY; without even the implied warranty of
- *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *      GNU Lesser General Public License for more details.
- *
- *      You should have received a copy of the GNU Lesser General Public License
- *      along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-package net.openhft.chronicle.hash.impl.stage.query;
-
-import net.openhft.chronicle.hash.impl.stage.entry.HashLookupPos;
-import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch;
-import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
-import net.openhft.sg.StageRef;
-import net.openhft.sg.Staged;
-
-@Staged
-public abstract class QueryHashLookupPos extends HashLookupPos {
-    
-    @StageRef SegmentStages s;
-    @StageRef HashLookupSearch hashLookupSearch;
-    
-    public void initHashLookupPos() {
-        // validation + make hashLookupPos a dependant of segmentTier. This is needed, because
-        // after tier change should re-perform hashLookupSearch, starting from the searchStartPos
-        int segmentTier = s.segmentTier;
-        assert segmentTier >= 0 : "segment tier " + segmentTier;
-        s.innerReadLock.lock();
-        this.hashLookupPos = hashLookupSearch.searchStartPos;
-    }
-}
@@ -54,7 +54,7 @@ import net.openhft.sg.Staged;
        KeySearch.class,
        InputKeyHashCode.class,
        QueryHashLookupSearch.class,
-        QueryHashLookupPos.class,
+        HashLookupPos.class,

        QueryCheckOnEachPublicOperation.class,
        SearchAllocatedChunks.class,
@@ -58,7 +58,7 @@ import net.openhft.sg.Staged;
        KeySearch.class,
        InputKeyHashCode.class,
        QueryHashLookupSearch.class,
-        QueryHashLookupPos.class,
+        HashLookupPos.class,

        QueryCheckOnEachPublicOperation.class,
        SearchAllocatedChunks.class,
@@ -201,7 +201,8 @@ public abstract class MapEntryStages<K, V> extends HashEntryStages<K>

        if (tierHasChanged) {
            // implicitly inits key search, locating hashLookupPos on the empty slot
-            assert ks.searchStateAbsent();
+            if (!ks.searchStateAbsent())
+                throw new AssertionError();
        }

        initValue(newValue);