From ff99f5cccd8a318c6c758a85f88814a7a6c8c72e Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 3 Aug 2015 16:38:35 +0300 Subject: [PATCH] On the way to HCOLL-360: Rename HashLookup to CompactOffHeapLinearHashTable and make it not a stage; extracted KeySearch from HashQuery --- ...ava => CompactOffHeapLinearHashTable.java} | 106 +++++------------- .../hash/impl/VanillaChronicleHash.java | 17 ++- .../impl/stage/entry/HashEntryStages.java | 2 +- .../hash/impl/stage/entry/HashLookupPos.java | 12 ++ .../hash/impl/stage/entry/SegmentStages.java | 19 ++-- .../impl/stage/iter/HashSegmentIteration.java | 14 ++- .../impl/stage/query/HashLookupSearch.java | 60 +++++----- .../hash/impl/stage/query/HashQuery.java | 80 ++----------- .../hash/impl/stage/query/KeySearch.java | 95 ++++++++++++++++ .../stage/query/SearchAllocatedChunks.java | 8 +- .../chronicle/map/ChronicleMapBuilder.java | 18 +-- .../map/impl/MapIterationContext.java | 1 - .../chronicle/map/impl/MapQueryContext.java | 9 +- .../impl/ReplicatedMapIterationContext.java | 1 - .../map/impl/ReplicatedMapQueryContext.java | 9 +- .../map/impl/stage/entry/MapEntryStages.java | 11 +- .../map/impl/stage/input/ReplicatedInput.java | 6 +- .../iter/ReplicatedMapSegmentIteration.java | 8 +- .../map/impl/stage/query/MapAbsent.java | 16 +-- .../map/impl/stage/query/MapQuery.java | 10 +- .../impl/stage/query/QueryMapEntryStages.java | 3 +- .../impl/stage/query/ReplicatedMapAbsent.java | 7 +- .../impl/stage/query/ReplicatedMapQuery.java | 2 +- 23 files changed, 255 insertions(+), 259 deletions(-) rename src/main/java/net/openhft/chronicle/hash/impl/{stage/entry/HashLookup.java => CompactOffHeapLinearHashTable.java} (61%) create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookup.java b/src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java similarity index 61% rename from src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookup.java rename to src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java index 1445af585..cc524e54a 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookup.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java @@ -14,20 +14,15 @@ * along with this program. If not, see . */ -package net.openhft.chronicle.hash.impl.stage.entry; +package net.openhft.chronicle.hash.impl; -import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.lang.Maths; -import net.openhft.sg.Stage; -import net.openhft.sg.StageRef; -import net.openhft.sg.Staged; import static net.openhft.lang.MemoryUnit.BITS; import static net.openhft.lang.MemoryUnit.BYTES; import static net.openhft.lang.io.NativeBytes.UNSAFE; -@Staged -public class HashLookup { +public class CompactOffHeapLinearHashTable { // to fit 64 bits per slot. public static final int MAX_SEGMENT_CHUNKS = 1 << 30; public static final int MAX_SEGMENT_ENTRIES = 1 << 29; @@ -71,25 +66,16 @@ public class HashLookup { public static final long UNSET_KEY = 0L; public static final long UNSET_ENTRY = 0L; - @Stage("SegmentHashLookup") long address = -1; - @Stage("SegmentHashLookup") long capacityMask; - @Stage("SegmentHashLookup") int hashLookupEntrySize; - @Stage("SegmentHashLookup") long capacityMask2; - @Stage("SegmentHashLookup") int keyBits; - @Stage("SegmentHashLookup") long keyMask; - @Stage("SegmentHashLookup") long valueMask; - @Stage("SegmentHashLookup") long entryMask; - public void initSegmentHashLookup( - long address, long capacity, int entrySize, int keyBits, int valueBits) { - innerInitSegmentHashLookup(address, capacity, entrySize, keyBits, valueBits); - } - - @Stage("SegmentHashLookup") - private void innerInitSegmentHashLookup( - long address, long capacity, int entrySize, int keyBits, int valueBits) { - this.address = address; + private final long capacityMask; + private final int hashLookupEntrySize; + private final long capacityMask2; + private final int keyBits; + private final long keyMask; + private final long valueMask; + private final long entryMask; + CompactOffHeapLinearHashTable(long capacity, int entrySize, int keyBits, int valueBits) { this.capacityMask = capacity - 1L; this.hashLookupEntrySize = entrySize; @@ -100,16 +86,10 @@ public class HashLookup { this.valueMask = mask(valueBits); this.entryMask = mask(keyBits + valueBits); } - - @StageRef VanillaChronicleHashHolder hh; - @StageRef SegmentStages s; - public void initSegmentHashLookup() { - long hashLookupOffset = hh.h().segmentOffset(s.segmentIndex); - innerInitSegmentHashLookup(hh.h().ms.address() + hashLookupOffset, - hh.h().segmentHashLookupCapacity, hh.h().segmentHashLookupEntrySize, - hh.h().segmentHashLookupKeyBits, hh.h().segmentHashLookupValueBits); - + CompactOffHeapLinearHashTable(VanillaChronicleHash h) { + this(h.segmentHashLookupCapacity, h.segmentHashLookupEntrySize, h.segmentHashLookupKeyBits, + h.segmentHashLookupValueBits); } long indexToPos(long index) { @@ -152,49 +132,44 @@ public class HashLookup { return (pos -= hashLookupEntrySize) >= 0 ? pos : capacityMask2; } - public long readEntry(long pos) { - return UNSAFE.getLong(address + pos); + public long readEntry(long addr, long pos) { + return UNSAFE.getLong(addr + pos); } - public void writeEntry(long pos, long prevEntry, long key, long value) { + public void writeEntryVolatile(long addr, long pos, long prevEntry, long key, long value) { long entry = (prevEntry & ~entryMask) | entry(key, value); - UNSAFE.putLong(address + pos, entry); - } - - public void writeEntryVolatile(long pos, long prevEntry, long key, long value) { - long entry = (prevEntry & ~entryMask) | entry(key, value); - UNSAFE.putLongVolatile(null, address + pos, entry); + UNSAFE.putLongVolatile(null, addr + pos, entry); } - public void putValueVolatile(long pos, long value) { + public void putValueVolatile(long addr, long pos, long value) { checkValueForPut(value); - long currentEntry = readEntry(pos); - writeEntryVolatile(pos, currentEntry, key(currentEntry), value); + long currentEntry = readEntry(addr, pos); + writeEntryVolatile(addr, pos, currentEntry, key(currentEntry), value); } - void writeEntry(long pos, long prevEntry, long anotherEntry) { + void writeEntry(long addr, long pos, long prevEntry, long anotherEntry) { long entry = (prevEntry & ~entryMask) | (anotherEntry & entryMask); - UNSAFE.putLong(address + pos, entry); + UNSAFE.putLong(addr + pos, entry); } - void clearEntry(long pos, long prevEntry) { + void clearEntry(long addr, long pos, long prevEntry) { long entry = (prevEntry & ~entryMask); - UNSAFE.putLong(address + pos, entry); + UNSAFE.putLong(addr + pos, entry); } - public void clearHashLookup() { - UNSAFE.setMemory(address, capacityMask2 + hashLookupEntrySize, (byte) 0); + public void clearHashLookup(long addr) { + UNSAFE.setMemory(addr, capacityMask2 + hashLookupEntrySize, (byte) 0); } /** * Returns "insert" position in terms of consequent putValue() */ - public long remove(long posToRemove) { - long entryToRemove = readEntry(posToRemove); + public long remove(long addr, long posToRemove) { + long entryToRemove = readEntry(addr, posToRemove); long posToShift = posToRemove; while (true) { posToShift = step(posToShift); - long entryToShift = readEntry(posToShift); + long entryToShift = readEntry(addr, posToShift); if (empty(entryToShift)) break; long insertPos = hlPos(key(entryToShift)); @@ -209,31 +184,12 @@ public class HashLookup { if ((cond1 && cond2) || // chain wrapped around capacity (posToShift < insertPos && (cond1 || cond2))) { - writeEntry(posToRemove, entryToRemove, entryToShift); + writeEntry(addr, posToRemove, entryToRemove, entryToShift); posToRemove = posToShift; entryToRemove = entryToShift; } } - clearEntry(posToRemove, entryToRemove); + clearEntry(addr, posToRemove, entryToRemove); return posToRemove; } - - interface EntryConsumer { - void accept(long key, long value); - } - - String hashLookupToString() { - final StringBuilder sb = new StringBuilder("{"); - forEach((key, value) -> sb.append(key).append('=').append(value).append(',')); - sb.append('}'); - return sb.toString(); - } - - void forEach(EntryConsumer action) { - for (long pos = 0L; pos <= capacityMask2; pos += hashLookupEntrySize) { - long entry = readEntry(pos); - if (!empty(entry)) - action.accept(key(entry), value(entry)); - } - } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java index f1aaa1692..604c935ab 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -20,7 +20,6 @@ import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.ExternalHashQueryContext; import net.openhft.chronicle.hash.HashEntry; import net.openhft.chronicle.hash.HashSegmentContext; -import net.openhft.chronicle.hash.impl.stage.entry.HashLookup; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.SizeMarshaller; @@ -40,9 +39,13 @@ import java.io.IOException; import java.io.ObjectInputStream; import java.io.Serializable; import java.nio.channels.FileChannel; +import java.util.List; import static java.lang.Long.numberOfTrailingZeros; import static java.lang.Math.max; +import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.entrySize; +import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.keyBits; +import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.valueBits; import static net.openhft.lang.MemoryUnit.*; public abstract class VanillaChronicleHash, @@ -107,6 +110,8 @@ public abstract class VanillaChronicleHash builder, boolean replicated) { @@ -134,12 +139,11 @@ public abstract class VanillaChronicleHash implements HashEntry { - @StageRef VanillaChronicleHashHolder hh; + @StageRef public VanillaChronicleHashHolder hh; @StageRef public SegmentStages s; @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef public HashLookupPos hlp; 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 89db0824b..6f4fab74c 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 @@ -16,7 +16,10 @@ package net.openhft.chronicle.hash.impl.stage.entry; +import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; +import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.sg.Stage; +import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @Staged @@ -36,4 +39,13 @@ public abstract class HashLookupPos { } public abstract void closeHashLookupPos(); + + @StageRef VanillaChronicleHashHolder hh; + @StageRef SegmentStages s; + + public void putValueVolatile(long newValue) { + CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; + hashLookup.checkValueForPut(newValue); + hashLookup.putValueVolatile(s.segmentBase, hashLookupPos, newValue); + } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java index 174a6f825..1cc549e63 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java @@ -21,7 +21,7 @@ import net.openhft.chronicle.hash.SegmentLock; import net.openhft.chronicle.hash.impl.*; import net.openhft.chronicle.hash.impl.stage.hash.Chaining; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; -import net.openhft.chronicle.hash.impl.stage.query.HashQuery; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.locks.InterProcessLock; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.collection.SingleThreadedDirectBitSet; @@ -269,8 +269,8 @@ public abstract class SegmentStages implements SegmentLock { private void linkToSegmentContextsChain() { SegmentStages innermostContextOnThisSegment = rootContextOnThisSegment; while (true) { - Data key = ((HashQuery) (Object) innermostContextOnThisSegment).inputKey; - if (Objects.equals(key, ((HashQuery) (Object) this).inputKey)) { + Data key = ((KeySearch) (Object) innermostContextOnThisSegment).inputKey; + if (Objects.equals(key, ((KeySearch) (Object) this).inputKey)) { throw new IllegalStateException("Nested same-thread contexts cannot access " + "the same key " + key); } @@ -324,14 +324,10 @@ public abstract class SegmentStages implements SegmentLock { localLockState = newState; } - // TODO their isHeldByCurrentThread should be _context_ local - // i. e. if outer same-thread context hold the lock, isHeld() should be false, but - // lock() op could be shallow @StageRef public ReadLock innerReadLock; @StageRef public UpdateLock innerUpdateLock; @StageRef public WriteLock innerWriteLock; - @NotNull @Override public InterProcessLock readLock() { @@ -352,14 +348,12 @@ public abstract class SegmentStages implements SegmentLock { checkOnEachPublicOperation.checkOnEachPublicOperation(); return innerWriteLock; } - - - @StageRef public HashLookup hashLookup; @Stage("Segment") MultiStoreBytes freeListBytes = new MultiStoreBytes(); @Stage("Segment") public SingleThreadedDirectBitSet freeList = new SingleThreadedDirectBitSet(); @Stage("Segment") long entrySpaceOffset = 0; - + @Stage("Segment") public long segmentBase; + boolean segmentInit() { return entrySpaceOffset > 0; } @@ -367,6 +361,7 @@ public abstract class SegmentStages implements SegmentLock { void initSegment() { VanillaChronicleHash h = hh.h(); long hashLookupOffset = h.segmentOffset(segmentIndex); + segmentBase = hh.h().ms.address() + hashLookupOffset; long freeListOffset = hashLookupOffset + h.segmentHashLookupOuterSize; freeListBytes.storePositionAndSize(h.ms, freeListOffset, h.segmentFreeListInnerSize); freeList.reuse(freeListBytes); @@ -432,7 +427,7 @@ public abstract class SegmentStages implements SegmentLock { public void clearSegment() { innerWriteLock.lock(); - hashLookup.clearHashLookup(); + hh.h().hashLookup.clearHashLookup(segmentBase); freeList.clear(); nextPosToSearchFrom(0L); entries(0L); 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 4a44384d8..8e2316f8d 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 @@ -17,8 +17,9 @@ package net.openhft.chronicle.hash.impl.stage.iter; import net.openhft.chronicle.hash.HashEntry; +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.HashLookup; 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; @@ -33,7 +34,7 @@ public abstract class HashSegmentIteration> implements @StageRef public SegmentStages s; @StageRef HashEntryStages e; - @StageRef protected HashLookup hashLookup; + @StageRef VanillaChronicleHashHolder hh; @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef protected HashLookupPos hlp; @@ -57,13 +58,14 @@ public abstract class HashSegmentIteration> implements return true; boolean interrupted = false; long startPos = 0L; - while (!hashLookup.empty(hashLookup.readEntry(startPos))) { + CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; + while (!hashLookup.empty(hashLookup.readEntry(s.segmentBase, startPos))) { startPos = hashLookup.step(startPos); } hlp.initHashLookupPos(startPos); do { hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos)); - long entry = hashLookup.readEntry(hlp.hashLookupPos); + long entry = hashLookup.readEntry(s.segmentBase, hlp.hashLookupPos); if (!hashLookup.empty(entry)) { e.readExistingEntry(hashLookup.value(entry)); if (entryIsPresent()) { @@ -105,10 +107,10 @@ public abstract class HashSegmentIteration> implements s.innerWriteLock.lock(); try { // this condition mean -- some other entry taken place of the removed one - if (hashLookup.remove(hlp.hashLookupPos) != hlp.hashLookupPos) { + if (hh.h().hashLookup.remove(s.segmentBase, hlp.hashLookupPos) != hlp.hashLookupPos) { // if so, should make step back, to compensate step forward on the next iteration, // to consume the shifted entry - hlp.setHashLookupPos(hashLookup.stepBack(hlp.hashLookupPos)); + hlp.setHashLookupPos(hh.h().hashLookup.stepBack(hlp.hashLookupPos)); } e.innerRemoveEntryExceptHashLookupUpdate(); } finally { diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java index 71fa9783d..129613643 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java @@ -16,49 +16,57 @@ package net.openhft.chronicle.hash.impl.stage.query; +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.HashLookup; import net.openhft.chronicle.hash.impl.stage.entry.HashLookupPos; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; -import static net.openhft.chronicle.hash.impl.stage.entry.HashLookup.UNSET_KEY; +import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.UNSET_KEY; @Staged public abstract class HashLookupSearch { @StageRef SegmentStages s; @StageRef HashEntryStages e; - @StageRef HashLookup hl; @StageRef HashQuery op; @StageRef VanillaChronicleHashHolder hh; @StageRef HashLookupPos hlp; + @StageRef KeySearch ks; @Stage("SearchKey") long searchKey = UNSET_KEY; @Stage("SearchKey") long searchStartPos; + + private CompactOffHeapLinearHashTable hl() { + return hh.h().hashLookup; + } void initSearchKey() { - searchKey = hl.maskUnsetKey(hh.h().hashSplitting.segmentHash(op.hashOfKey)); - searchStartPos = hl.hlPos(searchKey); + searchKey = hl().maskUnsetKey(hh.h().hashSplitting.segmentHash(op.hashOfKey)); + searchStartPos = hl().hlPos(searchKey); + } + + private long addr() { + return s.segmentBase; } public long nextPos() { long pos = hlp.hashLookupPos; while (true) { - long entry = hl.readEntry(pos); - if (hl.empty(entry)) { + long entry = hl().readEntry(addr(), pos); + if (hl().empty(entry)) { hlp.setHashLookupPos(pos); return -1L; } - pos = hl.step(pos); + pos = hl().step(pos); if (pos == searchStartPos) break; - if (hl.key(entry) == searchKey) { + if (hl().key(entry) == searchKey) { hlp.setHashLookupPos(pos); - return hl.value(entry); + return hl().value(entry); } } throw new IllegalStateException("MultiMap is full, that most likely means you " + @@ -67,39 +75,21 @@ public abstract class HashLookupSearch { } public void found() { - hlp.setHashLookupPos(hl.stepBack(hlp.hashLookupPos)); + hlp.setHashLookupPos(hl().stepBack(hlp.hashLookupPos)); } public void remove() { - hlp.setHashLookupPos(hl.remove(hlp.hashLookupPos)); + hlp.setHashLookupPos(hl().remove(addr(), hlp.hashLookupPos)); } - void put(long value) { - hl.checkValueForPut(value); - hl.writeEntry(hlp.hashLookupPos, hl.readEntry(hlp.hashLookupPos), searchKey, value); - } - - public void putVolatile(long value) { - hl.checkValueForPut(value); - long currentEntry = hl.readEntry(hlp.hashLookupPos); - assert hl.key(currentEntry) == searchKey; - hl.writeEntryVolatile(hlp.hashLookupPos, currentEntry, searchKey, value); - } - public void putNewVolatile(long value) { - hl.checkValueForPut(value); - long currentEntry = hl.readEntry(hlp.hashLookupPos); - hl.writeEntryVolatile(hlp.hashLookupPos, currentEntry, searchKey, value); + hl().checkValueForPut(value); + long currentEntry = hl().readEntry(addr(), hlp.hashLookupPos); + hl().writeEntryVolatile(addr(), hlp.hashLookupPos, currentEntry, searchKey, value); } public boolean checkSlotContainsExpectedKeyAndValue(long value) { - long entry = hl.readEntry(hlp.hashLookupPos); - return hl.key(entry) == searchKey && hl.value(entry) == value; + long entry = hl().readEntry(addr(), hlp.hashLookupPos); + return hl().key(entry) == searchKey && hl().value(entry) == value; } - - public boolean checkSlotIsEmpty() { - return hl.empty(hl.readEntry(hlp.hashLookupPos)); - } - - } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java index 808f39f5c..ea213830a 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java @@ -17,7 +17,6 @@ package net.openhft.chronicle.hash.impl.stage.query; import net.openhft.chronicle.algo.hashing.LongHashFunction; -import net.openhft.chronicle.bytes.BytesUtil; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.HashEntry; import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; @@ -27,12 +26,11 @@ 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.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; -import static net.openhft.chronicle.hash.impl.stage.query.HashQuery.SearchState.DELETED; -import static net.openhft.chronicle.hash.impl.stage.query.HashQuery.SearchState.PRESENT; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.DELETED; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged public abstract class HashQuery implements HashEntry { @@ -43,6 +41,7 @@ public abstract class HashQuery implements HashEntry { @StageRef public HashLookupSearch hashLookupSearch; @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef public HashLookupPos hlp; + @StageRef public KeySearch ks; @StageRef public InputKeyInstanceData inputKeyInstanceValue; @StageRef public InputKeyBytesData inputKeyBytesValue; @@ -52,8 +51,8 @@ public abstract class HashQuery implements HashEntry { if (s.concurrentSameThreadContexts && s.rootContextOnThisSegment.latestSameThreadSegmentModCount != s.contextModCount) { - if (keySearchInit()) { - if (searchState == PRESENT) { + if (ks.keySearchInit()) { + if (ks.searchState == PRESENT) { if (!hashLookupSearch.checkSlotContainsExpectedKeyAndValue(entry.pos)) hlp.closeHashLookupPos(); } @@ -62,90 +61,31 @@ public abstract class HashQuery implements HashEntry { } } - public Data inputKey = null; - - public void initInputKey(Data inputKey) { - this.inputKey = inputKey; - } - public Data queriedKey() { checkOnEachPublicOperation.checkOnEachPublicOperation(); - return inputKey; + return ks.inputKey; } public long hashOfKey = 0; void initHashOfKey() { - hashOfKey = inputKey.hash(LongHashFunction.city_1_1()); - } - - public enum SearchState { - PRESENT, - DELETED, - ABSENT - } - - @Stage("KeySearch") protected SearchState searchState = null; - - abstract boolean keySearchInit(); - - @Stage("KeySearch") - public void setSearchState(SearchState newSearchState) { - this.searchState = newSearchState; - } - - void initKeySearch() { - for (long pos; (pos = hashLookupSearch.nextPos()) >= 0L;) { - entry.readExistingEntry(pos); - if (!keyEquals()) - continue; - hashLookupSearch.found(); - keyFound(); - return; - } - searchState = SearchState.ABSENT; - } - - boolean keyEquals() { - return inputKey.size() == entry.keySize && - BytesUtil.bytesEqual(entry.entryBS, entry.keyOffset, - inputKey.bytes(), inputKey.offset(), entry.keySize); - } - - @Stage("KeySearch") - void keyFound() { - searchState = PRESENT; - } - - abstract void closeKeySearch(); - - public boolean searchStatePresent() { - return searchState == PRESENT; - } - - public boolean searchStateDeleted() { - return searchState == DELETED && !s.concurrentSameThreadContexts && - s.innerUpdateLock.isHeldByCurrentThread(); - } - - public boolean searchStateAbsent() { - return !searchStatePresent() && !searchStateDeleted(); + hashOfKey = ks.inputKey.hash(LongHashFunction.city_1_1()); } public boolean entryPresent() { - return searchStatePresent(); + return ks.searchStatePresent(); } @Override public void doRemove() { checkOnEachPublicOperation.checkOnEachPublicOperation(); s.innerUpdateLock.lock(); - if (searchStatePresent()) { + if (entryPresent()) { // TODO optimize: if shift-deletion is trivial, updateLock.lock() s.innerWriteLock.lock(); hashLookupSearch.remove(); entry.innerRemoveEntryExceptHashLookupUpdate(); - setSearchState(DELETED); + ks.setSearchState(DELETED); } else { throw new IllegalStateException("Entry is absent when doRemove() is called"); } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java new file mode 100644 index 000000000..7d1f1f88f --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java @@ -0,0 +1,95 @@ +/* + * 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.bytes.BytesUtil; +import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; +import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; +import net.openhft.sg.Stage; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; + +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.DELETED; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; + +@Staged +public abstract class KeySearch { + + @StageRef public SegmentStages s; + @StageRef public HashLookupSearch hashLookupSearch; + @StageRef public HashEntryStages entry; + + public Data inputKey = null; + + public void initInputKey(Data inputKey) { + this.inputKey = inputKey; + } + + public enum SearchState { + PRESENT, + DELETED, + ABSENT + } + + @Stage("KeySearch") protected SearchState searchState = null; + + abstract boolean keySearchInit(); + + @Stage("KeySearch") + public void setSearchState(SearchState newSearchState) { + this.searchState = newSearchState; + } + + void initKeySearch() { + for (long pos; (pos = hashLookupSearch.nextPos()) >= 0L;) { + entry.readExistingEntry(pos); + if (!keyEquals()) + continue; + hashLookupSearch.found(); + keyFound(); + return; + } + searchState = SearchState.ABSENT; + } + + boolean keyEquals() { + return inputKey.size() == entry.keySize && + BytesUtil.bytesEqual(entry.entryBS, entry.keyOffset, + inputKey.bytes(), inputKey.offset(), entry.keySize); + } + + @Stage("KeySearch") + void keyFound() { + searchState = PRESENT; + } + + abstract void closeKeySearch(); + + public boolean searchStatePresent() { + return searchState == PRESENT; + } + + public boolean searchStateDeleted() { + return searchState == DELETED && !s.concurrentSameThreadContexts && + s.innerUpdateLock.isHeldByCurrentThread(); + } + + public boolean searchStateAbsent() { + return !searchStatePresent() && !searchStateDeleted(); + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java index 85036f9c9..be6946ee5 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java @@ -20,16 +20,16 @@ import net.openhft.chronicle.hash.impl.stage.entry.AllocatedChunks; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; -import static net.openhft.chronicle.hash.impl.stage.query.HashQuery.SearchState.PRESENT; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged public class SearchAllocatedChunks extends AllocatedChunks { - @StageRef HashQuery q; + @StageRef KeySearch ks; @Override public void incrementSegmentEntriesIfNeeded() { - if (q.searchState != PRESENT) { + if (ks.searchState != PRESENT) { // update the size before the store fence s.entries(s.entries() + 1L); } @@ -40,6 +40,6 @@ public class SearchAllocatedChunks extends AllocatedChunks { // call incrementSegmentEntriesIfNeeded() before entry.writeNewEntry(), because the latter // clears out searchState, and it performs the search again, but in inconsistent state incrementSegmentEntriesIfNeeded(); - entry.writeNewEntry(s.alloc(allocatedChunks), q.inputKey); + entry.writeNewEntry(s.alloc(allocatedChunks), ks.inputKey); } } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 505fb8bdc..a62e09d09 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; -import net.openhft.chronicle.hash.impl.stage.entry.HashLookup; import net.openhft.chronicle.hash.replication.*; import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter; @@ -55,6 +54,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import static java.lang.Math.ceil; import static java.lang.Math.round; +import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.*; import static net.openhft.chronicle.hash.impl.util.Objects.builderEquals; import static net.openhft.chronicle.map.DefaultSpi.mapEntryOperations; import static net.openhft.chronicle.map.DefaultSpi.mapRemoteOperations; @@ -762,15 +762,17 @@ public final class ChronicleMapBuilder implements if (actualChunksPerSegment > 0) return entriesPerSegment; double averageChunksPerEntry = averageChunksPerEntry(replicated); - if (entriesPerSegment * averageChunksPerEntry > HashLookup.MAX_SEGMENT_CHUNKS) + if (entriesPerSegment * averageChunksPerEntry > + MAX_SEGMENT_CHUNKS) throw new IllegalStateException("Max chunks per segment is " + - HashLookup.MAX_SEGMENT_CHUNKS + " configured entries() and " + + MAX_SEGMENT_CHUNKS + + " configured entries() and " + "actualSegments() so that there should be " + entriesPerSegment + " entries per segment, while average chunks per entry is " + averageChunksPerEntry); - if (entriesPerSegment > HashLookup.MAX_SEGMENT_ENTRIES) + if (entriesPerSegment > MAX_SEGMENT_ENTRIES) throw new IllegalStateException("shouldn't be more than " + - HashLookup.MAX_SEGMENT_ENTRIES + " entries per segment"); + MAX_SEGMENT_ENTRIES + " entries per segment"); return entriesPerSegment; } @@ -975,9 +977,9 @@ public final class ChronicleMapBuilder implements } private int hashLookupSlotBytes(long entriesPerSegment, boolean replicated) { - int valueBits = HashLookup.valueBits(chunksPerSegment(entriesPerSegment, replicated)); - int keyBits = HashLookup.keyBits(entriesPerSegment, valueBits); - return HashLookup.entrySize(keyBits, valueBits); + int valueBits = valueBits(chunksPerSegment(entriesPerSegment, replicated)); + int keyBits = keyBits(entriesPerSegment, valueBits); + return entrySize(keyBits, valueBits); } private long trySegments(long entriesPerSegment, int maxSegments, boolean replicated) { diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java index fd15607bb..71261b8bf 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java @@ -42,7 +42,6 @@ import net.openhft.sg.Staged; MapSegmentIteration.class, - HashLookup.class, KeyBytesInterop.class, SegmentStages.class, HashLookupPos.class, 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 631fefe1b..ab4c8bedf 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java @@ -19,7 +19,6 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData; import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData; import net.openhft.chronicle.hash.impl.stage.data.instance.InputKeyInstanceData; -import net.openhft.chronicle.hash.impl.stage.entry.HashLookup; import net.openhft.chronicle.hash.impl.stage.entry.ReadLock; import net.openhft.chronicle.hash.impl.stage.entry.UpdateLock; import net.openhft.chronicle.hash.impl.stage.entry.WriteLock; @@ -27,10 +26,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder; import net.openhft.chronicle.hash.impl.stage.input.HashInputBytes; -import net.openhft.chronicle.hash.impl.stage.query.HashLookupSearch; -import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupPos; -import net.openhft.chronicle.hash.impl.stage.query.QuerySegmentStages; -import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks; +import net.openhft.chronicle.hash.impl.stage.query.*; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; import net.openhft.chronicle.map.impl.stage.data.bytes.InputFirstValueBytesData; import net.openhft.chronicle.map.impl.stage.data.bytes.InputSecondValueBytesData; @@ -55,10 +51,9 @@ import net.openhft.sg.Staged; VanillaChronicleMapHolderImpl.class, - HashLookup.class, KeyBytesInterop.class, QuerySegmentStages.class, - + KeySearch.class, HashLookupSearch.class, QueryHashLookupPos.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java index e595ba1bc..665d1cc0d 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java @@ -44,7 +44,6 @@ import net.openhft.sg.Staged; ReplicatedMapSegmentIteration.class, - HashLookup.class, KeyBytesInterop.class, SegmentStages.class, HashLookupPos.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 ffea2cd07..f4afc1c16 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java @@ -19,7 +19,6 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData; import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData; import net.openhft.chronicle.hash.impl.stage.data.instance.InputKeyInstanceData; -import net.openhft.chronicle.hash.impl.stage.entry.HashLookup; import net.openhft.chronicle.hash.impl.stage.entry.ReadLock; import net.openhft.chronicle.hash.impl.stage.entry.UpdateLock; import net.openhft.chronicle.hash.impl.stage.entry.WriteLock; @@ -28,10 +27,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder; import net.openhft.chronicle.hash.impl.stage.input.HashInputBytes; -import net.openhft.chronicle.hash.impl.stage.query.HashLookupSearch; -import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupPos; -import net.openhft.chronicle.hash.impl.stage.query.QuerySegmentStages; -import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks; +import net.openhft.chronicle.hash.impl.stage.query.*; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.*; import net.openhft.chronicle.map.impl.stage.data.instance.InputValueInstanceData; @@ -61,10 +57,9 @@ import net.openhft.sg.Staged; ReplicatedChronicleMapHolderImpl.class, - HashLookup.class, KeyBytesInterop.class, QuerySegmentStages.class, - + KeySearch.class, HashLookupSearch.class, QueryHashLookupPos.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 6d9415308..7455c4506 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 @@ -19,7 +19,8 @@ package net.openhft.chronicle.map.impl.stage.entry; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.entry.AllocatedChunks; import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; -import net.openhft.chronicle.hash.impl.stage.entry.HashLookup; +import net.openhft.chronicle.hash.impl.stage.query.HashLookupSearch; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.VanillaChronicleMap; import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; @@ -35,8 +36,6 @@ public abstract class MapEntryStages extends HashEntryStages @StageRef public VanillaChronicleMapHolder mh; @StageRef public AllocatedChunks allocatedChunks; - @StageRef HashLookup hashLookup; - long countValueSizeOffset() { return keyEnd(); @@ -114,9 +113,9 @@ public abstract class MapEntryStages extends HashEntryStages public void writeValueAndPutPos(Data value) { initValue(value); freeExtraAllocatedChunks(); - hashLookup.putValueVolatile(hlp.hashLookupPos, pos); + hlp.putValueVolatile(pos); } - + public long newSizeOfEverythingBeforeValue(Data newValue) { return valueSizeOffset + mh.m().valueSizeMarshaller.sizeEncodingSize(newValue.size()) - keySizeOffset; @@ -184,7 +183,7 @@ public abstract class MapEntryStages extends HashEntryStages } else { writeValue(newValue); } - hashLookup.putValueVolatile(hlp.hashLookupPos, pos); + hlp.putValueVolatile(pos); } protected void relocation(Data newValue, long newSizeOfEverythingBeforeValue) { diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java index 9ce9d6bd2..73d7ecb2b 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java @@ -20,6 +20,7 @@ import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.JavaLangBytesReusableBytesStore; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.replication.RemoteOperationContext; import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; @@ -44,6 +45,7 @@ public abstract class ReplicatedInput @StageRef LogHolder lh; @StageRef ReplicatedInputKeyBytesData replicatedInputKeyBytesValue; @StageRef ReplicatedInputValueBytesData replicatedInputValueBytesValue; + @StageRef KeySearch ks; @StageRef ReplicatedMapQuery q; @StageRef SegmentStages s; @StageRef ReplicatedMapEntryStages e; @@ -107,7 +109,7 @@ public abstract class ReplicatedInput if (isDeleted) { if (debugEnabled) { lh.LOG.debug("READING FROM SOURCE - into local-id={}, remote={}, remove(key={})", - mh.m().identifier(), riId, q.inputKey); + mh.m().identifier(), riId, ks.inputKey); } mh.m().remoteOperations.remove(this); return; @@ -117,7 +119,7 @@ public abstract class ReplicatedInput if (debugEnabled) { message = String.format( "READING FROM SOURCE - into local-id=%d, remote-id=%d, put(key=%s,", - mh.m().identifier(), riId, q.inputKey); + mh.m().identifier(), riId, ks.inputKey); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java index 03745f26c..378c164df 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java @@ -17,6 +17,8 @@ package net.openhft.chronicle.map.impl.stage.iter; import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; +import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.map.impl.ReplicatedIterationContextInterface; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; @@ -31,6 +33,7 @@ import java.util.function.Consumer; public abstract class ReplicatedMapSegmentIteration extends MapSegmentIteration implements ReplicatedIterationContextInterface, ReplicableEntry { + @StageRef VanillaChronicleHashHolder hh; @StageRef ReplicatedMapEntryStages e; @StageRef ReplicationUpdate ru; @StageRef DummyValueZeroData dummyValue; @@ -60,13 +63,14 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI if (entries == 0) return; long startPos = 0L; - while (!hashLookup.empty(hashLookup.readEntry(startPos))) { + CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; + while (!hashLookup.empty(hashLookup.readEntry(s.segmentBase, startPos))) { startPos = hashLookup.step(startPos); } hlp.initHashLookupPos(startPos); do { hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos)); - long entry = hashLookup.readEntry(hlp.hashLookupPos); + long entry = hashLookup.readEntry(s.segmentBase, hlp.hashLookupPos); if (!hashLookup.empty(entry)) { e.readExistingEntry(hashLookup.value(entry)); action.accept(this); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java index ca880e45d..be73f6dd6 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java @@ -20,6 +20,7 @@ import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.hash.impl.stage.query.HashLookupSearch; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.MapAbsentEntry; import net.openhft.chronicle.map.MapContext; import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; @@ -28,11 +29,12 @@ import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; -import static net.openhft.chronicle.hash.impl.stage.query.HashQuery.SearchState.PRESENT; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged public class MapAbsent implements MapAbsentEntry { + @StageRef public KeySearch ks; @StageRef MapQuery q; @StageRef MapEntryStages e; @StageRef public HashLookupSearch hashLookupSearch; @@ -41,8 +43,8 @@ public class MapAbsent implements MapAbsentEntry { @StageRef VanillaChronicleMapHolder mh; void putEntry(Data value) { - assert q.searchStateAbsent(); - long entrySize = e.entrySize(q.inputKey.size(), value.size()); + assert ks.searchStateAbsent(); + long entrySize = e.entrySize(ks.inputKey.size(), value.size()); q.allocatedChunks.initEntryAndKey(entrySize); e.initValue(value); e.freeExtraAllocatedChunks(); @@ -60,20 +62,20 @@ public class MapAbsent implements MapAbsentEntry { @Override public Data absentKey() { checkOnEachPublicOperation.checkOnEachPublicOperation(); - return q.inputKey; + return ks.inputKey; } @Override public void doInsert(Data value) { q.putPrefix(); - if (!q.searchStatePresent()) { - if (q.searchStateDeleted()) { + if (!q.entryPresent()) { + if (ks.searchStateDeleted()) { e.putValueDeletedEntry(value); } else { putEntry(value); } s.incrementModCount(); - q.setSearchState(PRESENT); + ks.setSearchState(PRESENT); } else { throw new IllegalStateException( "Entry is present in the map when doInsert() is called"); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java index 294c83736..f8a0722a3 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.query.HashQuery; +import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks; import net.openhft.chronicle.map.ExternalMapQueryContext; import net.openhft.chronicle.map.MapAbsentEntry; @@ -37,7 +38,7 @@ import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; -import static net.openhft.chronicle.hash.impl.stage.query.HashQuery.SearchState.PRESENT; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged public abstract class MapQuery extends HashQuery @@ -47,6 +48,7 @@ public abstract class MapQuery extends HashQuery @StageRef VanillaChronicleMapHolder mh; @StageRef MapEntryStages e; @StageRef SearchAllocatedChunks allocatedChunks; + @StageRef KeySearch ks; @StageRef public AcquireHandle acquireHandle; @StageRef public InputValueInstanceData inputValueInstanceValue; @@ -79,17 +81,17 @@ public abstract class MapQuery extends HashQuery s.innerUpdateLock.lock(); boolean searchResultsNotTrusted = underUpdatedLockIsHeld || s.concurrentSameThreadContexts; - if (hlp.hashLookupPosInit() && searchStateAbsent() && searchResultsNotTrusted) + if (hlp.hashLookupPosInit() && ks.searchStateAbsent() && searchResultsNotTrusted) hlp.closeHashLookupPos(); } @Override public void doReplaceValue(Data newValue) { putPrefix(); - if (searchStatePresent()) { + if (entryPresent()) { e.innerDefaultReplaceValue(newValue); s.incrementModCount(); - setSearchState(PRESENT); + ks.setSearchState(PRESENT); } else { throw new IllegalStateException( "Entry is absent in the map when doReplaceValue() is called"); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java index f877566bb..45d0439de 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java @@ -62,7 +62,8 @@ public abstract class QueryMapEntryStages extends MapEntryStages { } long entrySize = innerEntrySize(newSizeOfEverythingBeforeValue, newValue.size()); if (newValueSizeIsDifferent) { - allocatedChunks.initEntryAndKeyCopying(entrySize, valueSizeOffset - entryStartOffset); + allocatedChunks.initEntryAndKeyCopying( + entrySize, valueSizeOffset - entryStartOffset); initValue(newValue); } else { long oldValueSizeOffset = valueSizeOffset; diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java index 0233456ea..0bc56d977 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java @@ -17,12 +17,13 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.hash.Data; -import net.openhft.chronicle.hash.impl.stage.query.HashQuery; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; + @Staged public class ReplicatedMapAbsent extends MapAbsent { @@ -34,9 +35,9 @@ public class ReplicatedMapAbsent extends MapAbsent { public void doInsert(Data value) { q.putPrefix(); if (!q.entryPresent()) { - if (!q.searchStatePresent()) { + if (!ks.searchStatePresent()) { putEntry(value); - q.setSearchState(HashQuery.SearchState.PRESENT); + ks.setSearchState(PRESENT); } else { e.innerDefaultReplaceValue(value); s.deleted(s.deleted() - 1); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java index 42310a423..e99809fec 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java @@ -46,7 +46,7 @@ public abstract class ReplicatedMapQuery extends MapQuery if (entryPresent()) { return null; } else { - if (!searchStatePresent()) { + if (!ks.searchStatePresent()) { return absentDelegating; } else { assert e.entryDeleted();