From 81535163b2f295cb6ece4370c2285168a76e080d Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 7 Oct 2015 16:43:37 +0300 Subject: [PATCH] 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() --- .../hash/impl/stage/entry/HashLookupPos.java | 10 ++++ .../impl/stage/iter/HashSegmentIteration.java | 34 +++++++++--- .../stage/iter/IterationSegmentStages.java | 3 +- .../impl/stage/query/QueryHashLookupPos.java | 39 ------------- .../chronicle/map/impl/MapQueryContext.java | 2 +- .../map/impl/ReplicatedMapQueryContext.java | 2 +- .../map/impl/stage/entry/MapEntryStages.java | 3 +- ...OverflowAllocationDuringIterationTest.java | 55 +++++++++++++++++++ 8 files changed, 96 insertions(+), 52 deletions(-) delete mode 100644 src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java create mode 100644 src/test/java/net/openhft/chronicle/map/OverflowAllocationDuringIterationTest.java diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java index db67e0266..94bca664e 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java @@ -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; } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java index 55552b6eb..86f4dd487 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java @@ -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> 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> } private long forEachTierWhile( - Predicate action, long size, int currentTier, long currentTierBaseAddr) { + Predicate 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> } 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; } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java index 5313334be..c9b2ee58a 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java @@ -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; } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java deleted file mode 100644 index c10a2c8fd..000000000 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java +++ /dev/null @@ -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 . - */ - -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; - } -} diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java index 896f9ff32..63d110fcc 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java @@ -54,7 +54,7 @@ import net.openhft.sg.Staged; KeySearch.class, InputKeyHashCode.class, QueryHashLookupSearch.class, - QueryHashLookupPos.class, + HashLookupPos.class, QueryCheckOnEachPublicOperation.class, SearchAllocatedChunks.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java index e728b8d65..3e6457c8d 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java @@ -58,7 +58,7 @@ import net.openhft.sg.Staged; KeySearch.class, InputKeyHashCode.class, QueryHashLookupSearch.class, - QueryHashLookupPos.class, + HashLookupPos.class, QueryCheckOnEachPublicOperation.class, SearchAllocatedChunks.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java index 7baad557c..74cb9b199 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java @@ -201,7 +201,8 @@ public abstract class MapEntryStages extends HashEntryStages if (tierHasChanged) { // implicitly inits key search, locating hashLookupPos on the empty slot - assert ks.searchStateAbsent(); + if (!ks.searchStateAbsent()) + throw new AssertionError(); } initValue(newValue); diff --git a/src/test/java/net/openhft/chronicle/map/OverflowAllocationDuringIterationTest.java b/src/test/java/net/openhft/chronicle/map/OverflowAllocationDuringIterationTest.java new file mode 100644 index 000000000..ec9c57237 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/OverflowAllocationDuringIterationTest.java @@ -0,0 +1,55 @@ +/* + * 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 . + */ + +package net.openhft.chronicle.map; + +import org.junit.Test; + +public class OverflowAllocationDuringIterationTest { + + @Test + public void testOverflowAllocationDuringIteration() { + int entries = 10000; + String x = "x"; + try (ChronicleMap map = ChronicleMapBuilder + .of(Integer.class, CharSequence.class) + .averageValue(x) + .entries(entries) + .maxBloatFactor(10.0) + .actualSegments(1) + .create()) { + for (int i = 0; i < entries; i++) { + map.put(i, x); + } + for (int i = 0; i < 3; i++) { + final String currentX = x; + map.forEachEntry(e -> { + String v = e.value().get().toString(); + if (!currentX.contentEquals(v)) { + throw new AssertionError(currentX + " != " + v + "<=" + e.key()); + } + e.doReplaceValue(e.context().wrapValueAsData(v + v)); + }); + x = x + x; + for (int j = 0; j < entries; j++) { + if (map.get(j) == null || !x.contentEquals(map.get(j))) { + throw new AssertionError(); + } + } + } + } + } +}