Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

On the way to HCOLL-360: Rename HashLookup to CompactOffHeapLinearHashTable and make it not a stage; extracted KeySearch from HashQuery

Browse Source
This commit is contained in:
leventov
2015-08-03 20:20:14 +03:00
parent 7c10ea41ab
commit ff99f5cccd
23 changed files with 255 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookup.java → src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java
+31 -75
View File
@@ -14,20 +14,15 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
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));
}
}
}
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+11 -6
View File
@@ -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<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
@@ -107,6 +110,8 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
public transient long headerSize;
transient long segmentHeadersOffset;
transient long segmentsOffset;
public transient CompactOffHeapLinearHashTable hashLookup;
@SuppressWarnings("deprecation")
public VanillaChronicleHash(ChronicleMapBuilder<K, ?> builder, boolean replicated) {
@@ -134,12 +139,11 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
// Precomputed offsets and sizes for fast Context init
segmentHeaderSize = builder.segmentHeaderSize(replicated);
segmentHashLookupValueBits = HashLookup.valueBits(actualChunksPerSegment);
segmentHashLookupKeyBits =
HashLookup.keyBits(entriesPerSegment, segmentHashLookupValueBits);
segmentHashLookupValueBits = valueBits(actualChunksPerSegment);
segmentHashLookupKeyBits = keyBits(entriesPerSegment, segmentHashLookupValueBits);
segmentHashLookupEntrySize =
HashLookup.entrySize(segmentHashLookupKeyBits, segmentHashLookupValueBits);
segmentHashLookupCapacity = HashLookup.capacityFor(entriesPerSegment);
entrySize(segmentHashLookupKeyBits, segmentHashLookupValueBits);
segmentHashLookupCapacity = CompactOffHeapLinearHashTable.capacityFor(entriesPerSegment);
segmentHashLookupInnerSize = segmentHashLookupCapacity * segmentHashLookupEntrySize;
segmentHashLookupOuterSize = CACHE_LINES.align(segmentHashLookupInnerSize, BYTES);
@@ -183,6 +187,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
private void ownInitTransients() {
keyReaderProvider = Provider.of((Class) originalKeyReader.getClass());
keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass());
hashLookup = new CompactOffHeapLinearHashTable(this);
}
public final void createMappedStoreAndSegments(BytesStore bytesStore) throws IOException {
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java
+1 -1
View File
@@ -33,7 +33,7 @@ import org.jetbrains.annotations.NotNull;
@Staged
public abstract class HashEntryStages<K> implements HashEntry<K> {
@StageRef VanillaChronicleHashHolder<?, ?, ?> hh;
@StageRef public VanillaChronicleHashHolder<?, ?, ?> hh;
@StageRef public SegmentStages s;
@StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef public HashLookupPos hlp;
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java
+12
View File
@@ -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);
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
+7 -12
View File
@@ -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);
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
+8 -6
View File
@@ -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<K, E extends HashEntry<K>> implements
@StageRef public SegmentStages s;
@StageRef HashEntryStages<K> e;
@StageRef protected HashLookup hashLookup;
@StageRef VanillaChronicleHashHolder<?, ?, ?> hh;
@StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef protected HashLookupPos hlp;
@@ -57,13 +58,14 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>> 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<K, E extends HashEntry<K>> 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 {
src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java
+25 -35
View File
@@ -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));
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java
+10 -70
View File
@@ -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<K> implements HashEntry<K> {
@@ -43,6 +41,7 @@ public abstract class HashQuery<K> implements HashEntry<K> {
@StageRef public HashLookupSearch hashLookupSearch;
@StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef public HashLookupPos hlp;
@StageRef public KeySearch ks;
@StageRef public InputKeyInstanceData<K, ?, ?> inputKeyInstanceValue;
@StageRef public InputKeyBytesData<K> inputKeyBytesValue;
@@ -52,8 +51,8 @@ public abstract class HashQuery<K> implements HashEntry<K> {
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<K> implements HashEntry<K> {
}
}
public Data<K> inputKey = null;
public void initInputKey(Data<K> inputKey) {
this.inputKey = inputKey;
}
public Data<K> 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");
}
src/main/java/net/openhft/chronicle/hash/impl/stage/query/KeySearch.java
+95
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<K> {
@StageRef public SegmentStages s;
@StageRef public HashLookupSearch hashLookupSearch;
@StageRef public HashEntryStages<K> entry;
public Data<K> inputKey = null;
public void initInputKey(Data<K> 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();
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java
+4 -4
View File
@@ -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);
}
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+10 -8
View File
@@ -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<K, V> 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<K, V> 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) {
src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java
-1
View File
@@ -42,7 +42,6 @@ import net.openhft.sg.Staged;
MapSegmentIteration.class,
HashLookup.class,
KeyBytesInterop.class,
SegmentStages.class,
HashLookupPos.class,
src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
+2 -7
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java
-1
View File
@@ -44,7 +44,6 @@ import net.openhft.sg.Staged;
ReplicatedMapSegmentIteration.class,
HashLookup.class,
KeyBytesInterop.class,
SegmentStages.class,
HashLookupPos.class,
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java
+2 -7
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java
+5 -6
View File
@@ -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<K, V> extends HashEntryStages<K>
@StageRef public VanillaChronicleMapHolder<?, ?, ?, ?, ?, ?, ?> mh;
@StageRef public AllocatedChunks allocatedChunks;
@StageRef HashLookup hashLookup;
long countValueSizeOffset() {
return keyEnd();
@@ -114,9 +113,9 @@ public abstract class MapEntryStages<K, V> extends HashEntryStages<K>
public void writeValueAndPutPos(Data<V> value) {
initValue(value);
freeExtraAllocatedChunks();
hashLookup.putValueVolatile(hlp.hashLookupPos, pos);
hlp.putValueVolatile(pos);
}
public long newSizeOfEverythingBeforeValue(Data<V> newValue) {
return valueSizeOffset + mh.m().valueSizeMarshaller.sizeEncodingSize(newValue.size()) -
keySizeOffset;
@@ -184,7 +183,7 @@ public abstract class MapEntryStages<K, V> extends HashEntryStages<K>
} else {
writeValue(newValue);
}
hashLookup.putValueVolatile(hlp.hashLookupPos, pos);
hlp.putValueVolatile(pos);
}
protected void relocation(Data<V> newValue, long newSizeOfEverythingBeforeValue) {
src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java
+4 -2
View File
@@ -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<K, V, R>
@StageRef LogHolder lh;
@StageRef ReplicatedInputKeyBytesData<K> replicatedInputKeyBytesValue;
@StageRef ReplicatedInputValueBytesData<V> replicatedInputValueBytesValue;
@StageRef KeySearch<?> ks;
@StageRef ReplicatedMapQuery<K, V, ?> q;
@StageRef SegmentStages s;
@StageRef ReplicatedMapEntryStages<K, V, ?> e;
@@ -107,7 +109,7 @@ public abstract class ReplicatedInput<K, V, R>
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<K, V, R>
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);
}
src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java
+6 -2
View File
@@ -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<K, V, R> extends MapSegmentIteration<K, V, R>
implements ReplicatedIterationContextInterface<K, V, R>, ReplicableEntry {
@StageRef VanillaChronicleHashHolder<?, ?, ?> hh;
@StageRef ReplicatedMapEntryStages<K, V, ?> e;
@StageRef ReplicationUpdate<K> ru;
@StageRef DummyValueZeroData<V> dummyValue;
@@ -60,13 +63,14 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> 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);
src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java
+9 -7
View File
@@ -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<K, V> implements MapAbsentEntry<K, V> {
@StageRef public KeySearch<K> ks;
@StageRef MapQuery<K, V, ?> q;
@StageRef MapEntryStages<K, V> e;
@StageRef public HashLookupSearch hashLookupSearch;
@@ -41,8 +43,8 @@ public class MapAbsent<K, V> implements MapAbsentEntry<K, V> {
@StageRef VanillaChronicleMapHolder<K, ?, ?, V, ?, ?, ?> mh;
void putEntry(Data<V> 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<K, V> implements MapAbsentEntry<K, V> {
@Override
public Data<K> absentKey() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return q.inputKey;
return ks.inputKey;
}
@Override
public void doInsert(Data<V> 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");
src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java
+6 -4
View File
@@ -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<K, V, R> extends HashQuery<K>
@@ -47,6 +48,7 @@ public abstract class MapQuery<K, V, R> extends HashQuery<K>
@StageRef VanillaChronicleMapHolder<K, ?, ?, V, ?, ?, R> mh;
@StageRef MapEntryStages<K, V> e;
@StageRef SearchAllocatedChunks allocatedChunks;
@StageRef KeySearch<K> ks;
@StageRef public AcquireHandle<K, V> acquireHandle;
@StageRef public InputValueInstanceData<V, ?, ?> inputValueInstanceValue;
@@ -79,17 +81,17 @@ public abstract class MapQuery<K, V, R> extends HashQuery<K>
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<V> 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");
src/main/java/net/openhft/chronicle/map/impl/stage/query/QueryMapEntryStages.java
+2 -1
View File
@@ -62,7 +62,8 @@ public abstract class QueryMapEntryStages<K, V> extends MapEntryStages<K, V> {
}
long entrySize = innerEntrySize(newSizeOfEverythingBeforeValue, newValue.size());
if (newValueSizeIsDifferent) {
allocatedChunks.initEntryAndKeyCopying(entrySize, valueSizeOffset - entryStartOffset);
allocatedChunks.initEntryAndKeyCopying(
entrySize, valueSizeOffset - entryStartOffset);
initValue(newValue);
} else {
long oldValueSizeOffset = valueSizeOffset;
src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java
+4 -3
View File
@@ -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<K, V> extends MapAbsent<K, V> {
@@ -34,9 +35,9 @@ public class ReplicatedMapAbsent<K, V> extends MapAbsent<K, V> {
public void doInsert(Data<V> 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);
src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java
+1 -1
View File
@@ -46,7 +46,7 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
if (entryPresent()) {
return null;
} else {
if (!searchStatePresent()) {
if (!ks.searchStatePresent()) {
return absentDelegating;
} else {
assert e.entryDeleted();