diff --git a/pom.xml b/pom.xml
index 1c6a35b3f..4b6065c6b 100755
--- a/pom.xml
+++ b/pom.xml
@@ -149,6 +149,12 @@
test
+
+ net.openhft
+ stage-compiler
+ 1.0
+ provided
+
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/EntryKeyBytesData.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/EntryKeyBytesData.java
new file mode 100644
index 000000000..c394919e1
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/EntryKeyBytesData.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.data.bytes;
+
+import net.openhft.chronicle.bytes.RandomDataInput;
+import net.openhft.chronicle.hash.AbstractData;
+import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages;
+import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
+import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
+import net.openhft.sg.Stage;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public class EntryKeyBytesData extends AbstractData {
+
+ @StageRef KeyBytesInterop ki;
+ @StageRef HashEntryStages entry;
+ @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
+
+
+ @Stage("CachedEntryKey") private K cachedEntryKey;
+ @Stage("CachedEntryKey") private boolean cachedEntryKeyRead = false;
+
+ private void initCachedEntryKey() {
+ cachedEntryKey = innerGetUsing(cachedEntryKey);
+ cachedEntryKeyRead = true;
+ }
+
+ @Override
+ public RandomDataInput bytes() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return entry.entryBS;
+ }
+
+ @Override
+ public long offset() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return entry.keyOffset;
+ }
+
+ @Override
+ public long size() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return entry.keySize;
+ }
+
+ @Override
+ public K get() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return cachedEntryKey;
+ }
+
+ @Override
+ public K getUsing(K usingKey) {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return innerGetUsing(usingKey);
+ }
+
+ private K innerGetUsing(K usingKey) {
+ entry.entryBytes.position(entry.keyOffset);
+ return ki.keyReader.read(entry.entryBytes, size(), usingKey);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/InputKeyBytesData.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/InputKeyBytesData.java
new file mode 100644
index 000000000..781db3a19
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/InputKeyBytesData.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.data.bytes;
+
+import net.openhft.chronicle.bytes.RandomDataInput;
+import net.openhft.chronicle.hash.AbstractData;
+import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
+import net.openhft.chronicle.hash.impl.stage.input.HashInputBytes;
+import net.openhft.lang.io.Bytes;
+import net.openhft.sg.Stage;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+
+@Staged
+public class InputKeyBytesData extends AbstractData {
+ @StageRef HashInputBytes in;
+ @StageRef KeyBytesInterop ki;
+
+ @Stage("CachedBytesInputKey") private K cachedBytesInputKey;
+ @Stage("CachedBytesInputKey") private boolean cachedBytesInputKeyRead = false;
+
+ private void initCachedBytesInputKey() {
+ cachedBytesInputKey = getUsing(cachedBytesInputKey);
+ cachedBytesInputKeyRead = true;
+ }
+
+ @Override
+ public RandomDataInput bytes() {
+ return in.inputStore;
+ }
+
+ @Override
+ public long offset() {
+ return in.inputKeyOffset;
+ }
+
+ @Override
+ public long size() {
+ return in.inputKeySize;
+ }
+
+ @Override
+ public K get() {
+ return cachedBytesInputKey;
+ }
+
+ @Override
+ public K getUsing(K usingKey) {
+ Bytes inputBytes = in.inputBytes;
+ inputBytes.position(in.inputKeyOffset);
+ return ki.keyReader.read(inputBytes, size(), usingKey);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/data/instance/InputKeyInstanceData.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/instance/InputKeyInstanceData.java
new file mode 100644
index 000000000..8d43e1db2
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/instance/InputKeyInstanceData.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.data.instance;
+
+import net.openhft.chronicle.hash.impl.CopyingInstanceData;
+import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
+import net.openhft.chronicle.hash.impl.value.instance.KeyInitableData;
+import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
+import net.openhft.lang.io.DirectBytes;
+import net.openhft.sg.Stage;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public abstract class InputKeyInstanceData>
+ extends CopyingInstanceData implements KeyInitableData {
+
+ @StageRef KeyBytesInterop ki;
+
+ private K key = null;
+
+ @Override
+ public void initKey(K key) {
+ this.key = key;
+ }
+
+ @Override
+ public K instance() {
+ return key;
+ }
+
+ @Stage("Buffer") private DirectBytes buffer;
+ @Stage("Buffer") private boolean marshalled = false;
+
+ private void initBuffer() {
+ MKI mki = ki.keyMetaInterop(key);
+ long size = mki.size(ki.keyInterop, key);
+ buffer = getBuffer(this.buffer, size);
+ mki.write(ki.keyInterop, buffer, key);
+ buffer.flip();
+ marshalled = true;
+ }
+
+ @Override
+ public K getUsing(K usingKey) {
+ buffer.position(0);
+ return ki.keyReader.read(buffer, buffer.limit(), usingKey);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java
new file mode 100644
index 000000000..d49981a3c
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public class AllocatedChunks {
+
+ @StageRef public VanillaChronicleHashHolder, ?, ?> hh;
+ @StageRef public SegmentStages s;
+ @StageRef public HashEntryStages> entry;
+
+ public int allocatedChunks = 0;
+
+ public void initAllocatedChunks(int allocatedChunks) {
+ this.allocatedChunks = allocatedChunks;
+ }
+
+ public void incrementSegmentEntriesIfNeeded() {
+ // don't increment
+ }
+
+ public void initEntryAndKeyCopying(long entrySize, long bytesToCopy) {
+ initAllocatedChunks(hh.h().inChunks(entrySize));
+ entry.copyExistingEntry(s.alloc(allocatedChunks), bytesToCopy);
+ incrementSegmentEntriesIfNeeded();
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java
new file mode 100644
index 000000000..9f1b8b646
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashEntryStages.java
@@ -0,0 +1,134 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.chronicle.bytes.BytesStore;
+import net.openhft.chronicle.bytes.NativeBytesStore;
+import net.openhft.chronicle.hash.Data;
+import net.openhft.chronicle.hash.HashEntry;
+import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
+import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData;
+import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
+import net.openhft.lang.io.Bytes;
+import net.openhft.sg.Stage;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+import org.jetbrains.annotations.NotNull;
+
+
+@Staged
+public abstract class HashEntryStages implements HashEntry {
+
+ @StageRef VanillaChronicleHashHolder, ?, ?> hh;
+ @StageRef public SegmentStages s;
+ @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
+ @StageRef public HashLookupPos hlp;
+
+ public final Bytes entryBytes = hh.h().ms.bytes();
+ public final BytesStore entryBS =
+ new NativeBytesStore<>(entryBytes.address(), entryBytes.capacity(), null, false);
+
+ public long pos = -1;
+
+ public void initPos(long pos) {
+ this.pos = pos;
+ }
+
+ public abstract void closePos();
+
+ @Stage("EntryOffset") public long keySizeOffset = -1;
+
+ public void initEntryOffset() {
+ keySizeOffset = s.entrySpaceOffset + pos * hh.h().chunkSize;
+ entryBytes.limit(entryBytes.capacity());
+ }
+
+ public long keySize = -1;
+
+ public void initKeySize(long keySize) {
+ this.keySize = keySize;
+ }
+
+ public long keyOffset = -1;
+
+ public void initKeyOffset(long keyOffset) {
+ this.keyOffset = keyOffset;
+ }
+
+ public void readExistingEntry(long pos) {
+ initPos(pos);
+ entryBytes.position(keySizeOffset);
+ initKeySize(hh.h().keySizeMarshaller.readSize(entryBytes));
+ initKeyOffset(entryBytes.position());
+ }
+
+ public void writeNewEntry(long pos, Data> key) {
+ initPos(pos);
+ initKeySize(key.size());
+ entryBytes.position(keySizeOffset);
+ hh.h().keySizeMarshaller.writeSize(entryBytes, keySize);
+ initKeyOffset(entryBytes.position());
+ key.writeTo(entryBS, keyOffset);
+ }
+
+ public void copyExistingEntry(long newPos, long bytesToCopy) {
+ long oldKeySizeOffset = keySizeOffset;
+ long oldKeyOffset = keyOffset;
+ initPos(newPos);
+ initKeyOffset(keySizeOffset + (oldKeyOffset - oldKeySizeOffset));
+ entryBS.write(keySizeOffset, entryBS, oldKeySizeOffset, bytesToCopy);
+ }
+
+ public long keyEnd() {
+ return keyOffset + keySize;
+ }
+
+ protected long entryEnd() {
+ return keyEnd();
+ }
+
+ long entrySize() {
+ return entryEnd() - keySizeOffset;
+ }
+
+ @StageRef EntryKeyBytesData entryKey;
+
+ @NotNull
+ @Override
+ public Data key() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return entryKey;
+ }
+
+ @Stage("TheEntrySizeInChunks") public int entrySizeInChunks = 0;
+
+ void initTheEntrySizeInChunks() {
+ entrySizeInChunks = hh.h().inChunks(entrySize());
+ }
+
+ public void initTheEntrySizeInChunks(int actuallyUsedChunks) {
+ entrySizeInChunks = actuallyUsedChunks;
+ }
+
+ public void innerRemoveEntryExceptHashLookupUpdate() {
+ s.free(pos, entrySizeInChunks);
+ s.entries(s.entries() - 1L);
+ s.incrementModCount();
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookup.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookup.java
new file mode 100644
index 000000000..f5b8dc769
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookup.java
@@ -0,0 +1,238 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
+import net.openhft.chronicle.hash.impl.hashlookup.EntryConsumer;
+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 {
+ // to fit 64 bits per slot.
+ public static final int MAX_SEGMENT_CHUNKS = 1 << 30;
+ public static final int MAX_SEGMENT_ENTRIES = 1 << 29;
+
+
+ public static int valueBits(long actualChunksPerSegment) {
+ return 64 - Long.numberOfLeadingZeros(actualChunksPerSegment - 1L);
+ }
+
+ public static int keyBits(long entriesPerSegment, int valueBits) {
+ // key hash cardinality is between 1.0 and 2.0 of key cardinality
+ int minKeyBits = 64 - Long.numberOfLeadingZeros(entriesPerSegment - 1L);
+ // This minimizes probability of hash collision => on search, additional cache line touch
+ // and key comparison, that is quite high cost.
+ // between 8.0 and 16.0
+ minKeyBits += 3;
+ int actualEntryBits = (int) BYTES.align((long) (minKeyBits + valueBits), BITS);
+ // devote the rest bits for key
+ return actualEntryBits - valueBits;
+ }
+
+ public static int entrySize(int keyBits, int valueBits) {
+ return (int) BYTES.alignAndConvert((long) (keyBits + valueBits), BITS);
+ }
+
+ public static long capacityFor(long entriesPerSegment) {
+ if (entriesPerSegment < 0L)
+ throw new IllegalArgumentException("entriesPerSegment should be positive");
+ long capacity = Maths.nextPower2(entriesPerSegment, 64L);
+ if (((double) entriesPerSegment) / (double) capacity > 2./3.) {
+ // hash lookup shouldn't be too dense
+ capacity <<= 1L;
+ }
+ return capacity;
+ }
+
+ public static long mask(int bits) {
+ return (1L << bits) - 1L;
+ }
+
+ 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;
+
+ this.capacityMask = capacity - 1L;
+
+ this.hashLookupEntrySize = entrySize;
+ this.capacityMask2 = capacityMask * entrySize;
+
+ this.keyBits = keyBits;
+ this.keyMask = mask(keyBits);
+ 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);
+
+ }
+
+ long indexToPos(long index) {
+ return index * hashLookupEntrySize;
+ }
+
+ public long maskUnsetKey(long key) {
+ return (key &= keyMask) != UNSET_KEY ? key : keyMask;
+ }
+
+ public void checkValueForPut(long value) {
+ assert (value & ~valueMask) == 0L : "Value out of range, was " + value;
+ }
+
+ public boolean empty(long entry) {
+ return (entry & entryMask) == UNSET_ENTRY;
+ }
+
+ public long key(long entry) {
+ return entry & keyMask;
+ }
+
+ public long value(long entry) {
+ return (entry >>> keyBits) & valueMask;
+ }
+
+ long entry(long key, long value) {
+ return key | (value << keyBits);
+ }
+
+ public long hlPos(long key) {
+ return indexToPos(key & capacityMask);
+ }
+
+ public long step(long pos) {
+ return (pos += hashLookupEntrySize) <= capacityMask2 ? pos : 0L;
+ }
+
+ public long stepBack(long pos) {
+ return (pos -= hashLookupEntrySize) >= 0 ? pos : capacityMask2;
+ }
+
+ public long readEntry(long pos) {
+ return UNSAFE.getLong(address + pos);
+ }
+
+ public void writeEntry(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);
+ }
+
+ public void putValueVolatile(long pos, long value) {
+ checkValueForPut(value);
+ long currentEntry = readEntry(pos);
+ writeEntryVolatile(pos, currentEntry, key(currentEntry), value);
+ }
+
+ void writeEntry(long pos, long prevEntry, long anotherEntry) {
+ long entry = (prevEntry & ~entryMask) | (anotherEntry & entryMask);
+ UNSAFE.putLong(address + pos, entry);
+ }
+
+ void clearEntry(long pos, long prevEntry) {
+ long entry = (prevEntry & ~entryMask);
+ UNSAFE.putLong(address + pos, entry);
+ }
+
+ public void clearHashLookup() {
+ UNSAFE.setMemory(address, capacityMask2 + hashLookupEntrySize, (byte) 0);
+ }
+
+ /**
+ * Returns "insert" position in terms of consequent putValue()
+ */
+ public long remove(long posToRemove) {
+ long entryToRemove = readEntry(posToRemove);
+ long posToShift = posToRemove;
+ while (true) {
+ posToShift = step(posToShift);
+ long entryToShift = readEntry(posToShift);
+ if (empty(entryToShift))
+ break;
+ long insertPos = hlPos(key(entryToShift));
+ // the following condition essentially means circular permutations
+ // of three (r = posToRemove, s = posToShift, i = insertPos)
+ // positions are accepted:
+ // [...i..r...s.] or
+ // [...r..s...i.] or
+ // [...s..i...r.]
+ boolean cond1 = insertPos <= posToRemove;
+ boolean cond2 = posToRemove <= posToShift;
+ if ((cond1 && cond2) ||
+ // chain wrapped around capacity
+ (posToShift < insertPos && (cond1 || cond2))) {
+ writeEntry(posToRemove, entryToRemove, entryToShift);
+ posToRemove = posToShift;
+ entryToRemove = entryToShift;
+ }
+ }
+ clearEntry(posToRemove, entryToRemove);
+ return posToRemove;
+ }
+
+ 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/stage/entry/HashLookupPos.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java
new file mode 100644
index 000000000..c1d3ddb68
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupPos.java
@@ -0,0 +1,23 @@
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.sg.Stage;
+import net.openhft.sg.Staged;
+
+@Staged
+public abstract class HashLookupPos {
+
+ public long hashLookupPos = -1;
+
+ public abstract boolean hashLookupPosInit();
+
+ public void initHashLookupPos(long hashLookupPos) {
+ this.hashLookupPos = hashLookupPos;
+ }
+
+ @Stage("HashLookupPos")
+ public void setHashLookupPos(long hashLookupPos) {
+ this.hashLookupPos = hashLookupPos;
+ }
+
+ public abstract void closeHashLookupPos();
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java
new file mode 100644
index 000000000..20bc158e4
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+import net.openhft.chronicle.hash.locks.InterProcessLock;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.TimeUnit;
+
+import static net.openhft.chronicle.hash.impl.LocalLockState.READ_LOCKED;
+import static net.openhft.chronicle.hash.impl.LocalLockState.UNLOCKED;
+
+
+@Staged
+public class ReadLock implements InterProcessLock {
+
+ @StageRef SegmentStages s;
+ @StageRef HashEntryStages entry;
+ @StageRef HashLookupPos hlp;
+
+ @Override
+ public boolean isHeldByCurrentThread() {
+ return s.localLockState.read;
+ }
+
+ @Override
+ public void lock() {
+ if (s.localLockState == UNLOCKED) {
+ s.segmentHeader.readLock(s.segmentHeaderAddress);
+ s.setLocalLockState(READ_LOCKED);
+ }
+ }
+
+ @Override
+ public void lockInterruptibly() throws InterruptedException {
+ if (s.localLockState == UNLOCKED) {
+ s.segmentHeader.readLockInterruptibly(s.segmentHeaderAddress);
+ s.setLocalLockState(READ_LOCKED);
+ }
+ }
+
+ @Override
+ public boolean tryLock() {
+ if (s.localLockState == UNLOCKED) {
+ if (s.segmentHeader.tryReadLock(s.segmentHeaderAddress)) {
+ s.setLocalLockState(READ_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
+ if (s.localLockState == UNLOCKED) {
+ if (s.segmentHeader.tryReadLock(s.segmentHeaderAddress, time, unit)) {
+ s.setLocalLockState(READ_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ public void unlock() {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ return;
+ case READ_LOCKED:
+ s.segmentHeader.readUnlock(s.segmentHeaderAddress);
+ break;
+ case UPDATE_LOCKED:
+ s.segmentHeader.updateUnlock(s.segmentHeaderAddress);
+ break;
+ case WRITE_LOCKED:
+ s.segmentHeader.writeUnlock(s.segmentHeaderAddress);
+ }
+ s.setLocalLockState(UNLOCKED);
+ // TODO what should close here?
+ hlp.closeHashLookupPos();
+ entry.closePos();
+ }
+}
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
new file mode 100644
index 000000000..b984320f5
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
@@ -0,0 +1,368 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.chronicle.hash.impl.BigSegmentHeader;
+import net.openhft.chronicle.hash.impl.LocalLockState;
+import net.openhft.chronicle.hash.impl.SegmentHeader;
+import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
+import net.openhft.chronicle.hash.impl.stage.hash.Chaining;
+import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
+import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
+import net.openhft.chronicle.hash.locks.InterProcessLock;
+import net.openhft.chronicle.hash.locks.InterProcessReadWriteUpdateLock;
+import net.openhft.lang.collection.DirectBitSet;
+import net.openhft.lang.collection.SingleThreadedDirectBitSet;
+import net.openhft.lang.io.MultiStoreBytes;
+import net.openhft.sg.Stage;
+import net.openhft.sg.Staged;
+import net.openhft.sg.StageRef;
+import org.jetbrains.annotations.NotNull;
+
+import static net.openhft.chronicle.hash.impl.LocalLockState.UNLOCKED;
+
+@Staged
+public abstract class SegmentStages implements InterProcessReadWriteUpdateLock {
+
+ @StageRef Chaining chaining;
+ @StageRef public VanillaChronicleHashHolder, ?, ?> hh;
+ @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
+
+ @Stage("TheSegmentIndex") public int segmentIndex = -1;
+
+ public void initTheSegmentIndex(int segmentIndex) {
+ this.segmentIndex = segmentIndex;
+
+ }
+
+ @Stage("SegHeader") long segmentHeaderAddress;
+ @Stage("SegHeader") SegmentHeader segmentHeader = null;
+
+ private void initSegHeader() {
+ segmentHeaderAddress = hh.h().ms.address() + hh.h().segmentHeaderOffset(segmentIndex);
+ segmentHeader = BigSegmentHeader.INSTANCE;
+ }
+
+ public long entries() {
+ return segmentHeader.size(segmentHeaderAddress);
+ }
+
+ public void entries(long size) {
+ segmentHeader.size(segmentHeaderAddress, size);
+ }
+
+ long nextPosToSearchFrom() {
+ return segmentHeader.nextPosToSearchFrom(segmentHeaderAddress);
+ }
+
+ public void nextPosToSearchFrom(long nextPosToSearchFrom) {
+ segmentHeader.nextPosToSearchFrom(segmentHeaderAddress, nextPosToSearchFrom);
+ }
+
+ public long deleted() {
+ return segmentHeader.deleted(segmentHeaderAddress);
+ }
+
+ public void deleted(long deleted) {
+ segmentHeader.deleted(segmentHeaderAddress, deleted);
+ }
+
+ public long size() {
+ return entries() - deleted();
+ }
+
+
+ @Stage("Locks") public SegmentStages rootContextOnThisSegment = null;
+ /**
+ * See the ChMap Ops spec, considerations of nested same-thread concurrent contexts.
+ * Once context enters the segment, and observes concurrent same-thread context,
+ * it sets concurrentSameThreadContexts = true for itself and that concurrent context.
+ * This flag is not dropped on exit of one of these contexts, because between calls of
+ * this context, nested one could be initialized, does some changes that break our thread-local
+ * assumptions _and exit_, that is why on exit concurrent context should remain "dirty".
+ */
+ @Stage("Locks") public boolean concurrentSameThreadContexts;
+
+ @Stage("Locks") public int latestSameThreadSegmentModCount;
+ @Stage("Locks") public int contextModCount;
+
+ @Stage("Locks")
+ public void incrementModCount() {
+ contextModCount = rootContextOnThisSegment.latestSameThreadSegmentModCount =
+ rootContextOnThisSegment.latestSameThreadSegmentModCount + 1;
+ }
+
+ // chain
+ @Stage("Locks") SegmentStages nextNode;
+
+ @Stage("Locks") LocalLockState localLockState;
+ @Stage("Locks") int totalReadLockCount;
+ @Stage("Locks") int totalUpdateLockCount;
+ @Stage("Locks") int totalWriteLockCount;
+
+ public abstract boolean locksInit();
+
+ void initLocks() {
+ localLockState = UNLOCKED;
+ int indexOfThisContext = chaining.indexInContextChain;
+ for (int i = indexOfThisContext - 1; i >= 0; i--) {
+ if (tryFindInitLocksOfThisSegment(this, i))
+ return;
+ }
+ for (int i = indexOfThisContext + 1, size = chaining.contextChain.size(); i < size; i++) {
+ if (tryFindInitLocksOfThisSegment(this, i))
+ return;
+ }
+ rootContextOnThisSegment = this;
+ concurrentSameThreadContexts = false;
+ latestSameThreadSegmentModCount = 0;
+ contextModCount = 0;
+ nextNode = null;
+ totalReadLockCount = 0;
+ totalUpdateLockCount = 0;
+ totalWriteLockCount = 0;
+ }
+
+ boolean tryFindInitLocksOfThisSegment(Object thisContext, int index) {
+ SegmentStages c = chaining.contextAtIndexInChain(index);
+ if (c.segmentHeader != null && c.segmentHeaderAddress == segmentHeaderAddress &&
+ c.rootContextOnThisSegment != null) {
+ //TODO implement nested contexts. Hardest part - bodies of Read/Update/WriteLock classes
+ throw new IllegalStateException("Nested context not implemented yet");
+ } else {
+ return false;
+ }
+ }
+
+ void closeLocks() {
+ // TODO should throw ISE on attempt to close root context when children not closed
+ if (rootContextOnThisSegment == this) {
+ closeRootLocks();
+ } else {
+ closeNestedLocks();
+ }
+ localLockState = null;
+ rootContextOnThisSegment = null;
+ }
+
+ @Stage("Locks")
+ private void closeNestedLocks() {
+ unlinkFromSegmentContextsChain();
+
+ switch (localLockState) {
+ case UNLOCKED:
+ break;
+ case READ_LOCKED:
+ int newTotalReadLockCount = this.rootContextOnThisSegment.totalReadLockCount -= 1;
+ if (newTotalReadLockCount == 0) {
+ if (this.rootContextOnThisSegment.totalUpdateLockCount == 0 &&
+ this.rootContextOnThisSegment.totalWriteLockCount == 0) {
+ segmentHeader.readUnlock(segmentHeaderAddress);
+ }
+ } else if (newTotalReadLockCount < 0) {
+ throw new IllegalStateException("read underflow");
+ }
+ break;
+ case UPDATE_LOCKED:
+ int newTotalUpdateLockCount =
+ this.rootContextOnThisSegment.totalUpdateLockCount -= 1;
+ if (newTotalUpdateLockCount == 0) {
+ if (this.rootContextOnThisSegment.totalWriteLockCount == 0) {
+ if (this.rootContextOnThisSegment.totalReadLockCount == 0) {
+ segmentHeader.updateUnlock(segmentHeaderAddress);
+ } else {
+ segmentHeader.downgradeUpdateToReadLock(segmentHeaderAddress);
+ }
+ }
+ } else if (newTotalUpdateLockCount < 0) {
+ throw new IllegalStateException("update underflow");
+ }
+ break;
+ case WRITE_LOCKED:
+ int newTotalWriteLockCount = this.rootContextOnThisSegment.totalWriteLockCount -= 1;
+ if (newTotalWriteLockCount == 0) {
+ if (this.rootContextOnThisSegment.totalUpdateLockCount > 0) {
+ segmentHeader.downgradeWriteToUpdateLock(segmentHeaderAddress);
+ } else {
+ if (this.rootContextOnThisSegment.totalReadLockCount > 0) {
+ segmentHeader.downgradeWriteToReadLock(segmentHeaderAddress);
+ } else {
+ segmentHeader.writeUnlock(segmentHeaderAddress);
+ }
+ }
+ }
+ break;
+ }
+ }
+
+ @Stage("Locks")
+ private void unlinkFromSegmentContextsChain() {
+ SegmentStages prevContext = this.rootContextOnThisSegment;
+ while (true) {
+ assert prevContext.nextNode != null;
+ if (prevContext.nextNode == this)
+ break;
+ prevContext = prevContext.nextNode;
+ }
+ assert nextNode == null;
+ prevContext.nextNode = null;
+ }
+
+ @Stage("Locks")
+ private void closeRootLocks() {
+ assert nextNode == null;
+
+
+ switch (localLockState) {
+ case UNLOCKED:
+ return;
+ case READ_LOCKED:
+ segmentHeader.readUnlock(segmentHeaderAddress);
+ return;
+ case UPDATE_LOCKED:
+ segmentHeader.updateUnlock(segmentHeaderAddress);
+ return;
+ case WRITE_LOCKED:
+ segmentHeader.writeUnlock(segmentHeaderAddress);
+ }
+ }
+
+ @Stage("Locks")
+ public void setLocalLockState(LocalLockState newState) {
+ 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() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return innerReadLock;
+ }
+
+ @NotNull
+ @Override
+ public InterProcessLock updateLock() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return innerUpdateLock;
+ }
+
+ @NotNull
+ @Override
+ public InterProcessLock writeLock() {
+ 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;
+
+ boolean segmentInit() {
+ return entrySpaceOffset > 0;
+ }
+
+ void initSegment() {
+ VanillaChronicleHash, ?, ?, ?, ?> h = hh.h();
+ long hashLookupOffset = h.segmentOffset(segmentIndex);
+ long freeListOffset = hashLookupOffset + h.segmentHashLookupOuterSize;
+ freeListBytes.storePositionAndSize(h.ms, freeListOffset, h.segmentFreeListInnerSize);
+ freeList.reuse(freeListBytes);
+ entrySpaceOffset = freeListOffset + h.segmentFreeListOuterSize +
+ h.segmentEntrySpaceInnerOffset;
+ }
+
+ void closeSegment() {
+ entrySpaceOffset = 0;
+ }
+
+ //TODO refactor/optimize
+ public long alloc(int chunks) {
+ VanillaChronicleHash, ?, ?, ?, ?> h = hh.h();
+ if (chunks > h.maxChunksPerEntry)
+ throw new IllegalArgumentException("Entry is too large: requires " + chunks +
+ " entry size chucks, " + h.maxChunksPerEntry + " is maximum.");
+ long ret = freeList.setNextNContinuousClearBits(nextPosToSearchFrom(), chunks);
+ if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) {
+ if (ret != DirectBitSet.NOT_FOUND &&
+ ret + chunks > h.actualChunksPerSegment && ret < h.actualChunksPerSegment)
+ freeList.clear(ret, h.actualChunksPerSegment);
+ ret = freeList.setNextNContinuousClearBits(0L, chunks);
+ if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) {
+ if (ret != DirectBitSet.NOT_FOUND &&
+ ret + chunks > h.actualChunksPerSegment &&
+ ret < h.actualChunksPerSegment)
+ freeList.clear(ret, h.actualChunksPerSegment);
+ if (chunks == 1) {
+ throw new IllegalStateException(
+ "Segment is full, no free entries found");
+ } else {
+ throw new IllegalStateException(
+ "Segment is full or has no ranges of " + chunks
+ + " continuous free chunks"
+ );
+ }
+ }
+ updateNextPosToSearchFrom(ret, chunks);
+ } else {
+ // if bit at nextPosToSearchFrom is clear, it was skipped because
+ // more than 1 chunk was requested. Don't move nextPosToSearchFrom
+ // in this case. chunks == 1 clause is just a fast path.
+ if (chunks == 1 || freeList.isSet(nextPosToSearchFrom())) {
+ updateNextPosToSearchFrom(ret, chunks);
+ }
+ }
+ return ret;
+ }
+
+ public void free(long fromPos, int chunks) {
+ freeList.clear(fromPos, fromPos + chunks);
+ if (fromPos < nextPosToSearchFrom())
+ nextPosToSearchFrom(fromPos);
+ }
+
+ public void updateNextPosToSearchFrom(long allocated, int chunks) {
+ long nextPosToSearchFrom = allocated + chunks;
+ if (nextPosToSearchFrom >= hh.h().actualChunksPerSegment)
+ nextPosToSearchFrom = 0L;
+ nextPosToSearchFrom(nextPosToSearchFrom);
+ }
+
+ public void clearSegment() {
+ innerWriteLock.lock();
+ hashLookup.clearHashLookup();
+ freeList.clear();
+ nextPosToSearchFrom(0L);
+ entries(0L);
+ }
+
+ public void clear() {
+ clearSegment();
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java
new file mode 100644
index 000000000..67cd5edfd
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java
@@ -0,0 +1,131 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+import net.openhft.chronicle.hash.locks.InterProcessLock;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.TimeUnit;
+
+import static net.openhft.chronicle.hash.impl.LocalLockState.READ_LOCKED;
+import static net.openhft.chronicle.hash.impl.LocalLockState.UPDATE_LOCKED;
+
+@Staged
+public class UpdateLock implements InterProcessLock {
+
+ @StageRef SegmentStages s;
+
+ @Override
+ public boolean isHeldByCurrentThread() {
+ return s.localLockState.update;
+ }
+
+ @Override
+ public void lock() {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ s.segmentHeader.updateLock(s.segmentHeaderAddress);
+ s.setLocalLockState(UPDATE_LOCKED);
+ return;
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ case WRITE_LOCKED:
+ // do nothing
+ }
+ }
+
+ /**
+ * Non-static because after compilation it becomes inner class which forbids static methods
+ */
+ @NotNull
+ private IllegalMonitorStateException forbiddenUpgrade() {
+ return new IllegalMonitorStateException("Cannot upgrade from read to update lock");
+ }
+
+ @Override
+ public void lockInterruptibly() throws InterruptedException {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ s.segmentHeader.updateLockInterruptibly(s.segmentHeaderAddress);
+ s.setLocalLockState(UPDATE_LOCKED);
+ return;
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ case WRITE_LOCKED:
+ // do nothing
+ }
+ }
+
+ @Override
+ public boolean tryLock() {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ if (s.segmentHeader.tryUpdateLock(s.segmentHeaderAddress)) {
+ s.setLocalLockState(UPDATE_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ case WRITE_LOCKED:
+ return true;
+ }
+ throw new AssertionError();
+ }
+
+ @Override
+ public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ if (s.segmentHeader.tryUpdateLock(s.segmentHeaderAddress, time, unit)) {
+ s.setLocalLockState(UPDATE_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ case WRITE_LOCKED:
+ return true;
+ }
+ throw new AssertionError();
+ }
+
+ @Override
+ public void unlock() {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ case READ_LOCKED:
+ return;
+ case UPDATE_LOCKED:
+ s.segmentHeader.downgradeUpdateToReadLock(s.segmentHeaderAddress);
+ break;
+ case WRITE_LOCKED:
+ s.segmentHeader.downgradeWriteToReadLock(s.segmentHeaderAddress);
+ }
+ s.setLocalLockState(READ_LOCKED);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java
new file mode 100644
index 000000000..b10e9f554
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java
@@ -0,0 +1,146 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+import net.openhft.chronicle.hash.locks.InterProcessLock;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.TimeUnit;
+
+import static net.openhft.chronicle.hash.impl.LocalLockState.UPDATE_LOCKED;
+import static net.openhft.chronicle.hash.impl.LocalLockState.WRITE_LOCKED;
+
+@Staged
+public class WriteLock implements InterProcessLock {
+
+ @StageRef SegmentStages s;
+
+ @Override
+ public boolean isHeldByCurrentThread() {
+ return s.localLockState.write;
+ }
+
+ @Override
+ public void lock() {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ s.segmentHeader.writeLock(s.segmentHeaderAddress);
+ s.setLocalLockState(WRITE_LOCKED);
+ return;
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ s.segmentHeader.upgradeUpdateToWriteLock(s.segmentHeaderAddress);
+ s.setLocalLockState(WRITE_LOCKED);
+ case WRITE_LOCKED:
+ // do nothing
+ }
+ }
+
+ /**
+ * Non-static because after compilation it becomes inner class which forbids static methods
+ */
+ @NotNull
+ private IllegalMonitorStateException forbiddenUpgrade() {
+ return new IllegalMonitorStateException("Cannot upgrade from read to write lock");
+ }
+
+ @Override
+ public void lockInterruptibly() throws InterruptedException {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ s.segmentHeader.writeLockInterruptibly(s.segmentHeaderAddress);
+ s.setLocalLockState(WRITE_LOCKED);
+ return;
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ s.segmentHeader.upgradeUpdateToWriteLockInterruptibly(s.segmentHeaderAddress);
+ s.setLocalLockState(WRITE_LOCKED);
+ case WRITE_LOCKED:
+ // do nothing
+ }
+ }
+
+ @Override
+ public boolean tryLock() {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ if (s.segmentHeader.tryWriteLock(s.segmentHeaderAddress)) {
+ s.setLocalLockState(WRITE_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ if (s.segmentHeader.tryUpgradeUpdateToWriteLock(s.segmentHeaderAddress)) {
+ s.setLocalLockState(WRITE_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ case WRITE_LOCKED:
+ return true;
+ }
+ throw new AssertionError();
+ }
+
+ @Override
+ public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ if (s.segmentHeader.tryWriteLock(s.segmentHeaderAddress, time, unit)) {
+ s.setLocalLockState(WRITE_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ case READ_LOCKED:
+ throw forbiddenUpgrade();
+ case UPDATE_LOCKED:
+ if (s.segmentHeader.tryUpgradeUpdateToWriteLock(
+ s.segmentHeaderAddress, time, unit)) {
+ s.setLocalLockState(WRITE_LOCKED);
+ return true;
+ } else {
+ return false;
+ }
+ case WRITE_LOCKED:
+ return true;
+ }
+ throw new AssertionError();
+ }
+
+ @Override
+ public void unlock() {
+ switch (s.localLockState) {
+ case UNLOCKED:
+ case READ_LOCKED:
+ case UPDATE_LOCKED:
+ return;
+ case WRITE_LOCKED:
+ s.segmentHeader.downgradeWriteToUpdateLock(s.segmentHeaderAddress);
+ }
+ s.setLocalLockState(UPDATE_LOCKED);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java
new file mode 100644
index 000000000..79274ec64
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.hash;
+
+import net.openhft.sg.Staged;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Staged
+public class Chaining {
+
+ public final List contextChain;
+ public final int indexInContextChain;
+
+ public Chaining() {
+ contextChain = new ArrayList<>();
+ contextChain.add(this);
+ indexInContextChain = 0;
+ }
+
+ public Chaining(Chaining root) {
+ contextChain = root.contextChain;
+ indexInContextChain = contextChain.size();
+ contextChain.add(this);
+ }
+
+ public T contextAtIndexInChain(int index) {
+ //noinspection unchecked
+ return (T) contextChain.get(index);
+ }
+
+
+ boolean used;
+
+ public boolean usedInit() {
+ return used;
+ }
+
+ public void initUsed(boolean used) {
+ this.used = used;
+ }
+
+ void closeUsed() {
+ used = false;
+ }
+
+ public T getContext() {
+ for (Chaining context : contextChain) {
+ if (!context.usedInit()) {
+ //noinspection unchecked
+ return (T) context;
+ }
+ }
+ int maxNestedContexts = 1 << 16;
+ if (contextChain.size() > maxNestedContexts) {
+ throw new IllegalStateException("More than " + maxNestedContexts +
+ " nested ChronicleHash contexts are not supported. Very probable that " +
+ "you simply forgot to close context somewhere (recommended to use " +
+ "try-with-resources statement). " +
+ "Otherwise this is a bug, please report with this " +
+ "stack trace on https://github.com/OpenHFT/Chronicle-Map/issues");
+ }
+ //noinspection unchecked
+ return (T) createChaining();
+ }
+
+ public Chaining createChaining() {
+ return new Chaining(this);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java
new file mode 100644
index 000000000..d40e582bf
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.hash;
+
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public abstract class CheckOnEachPublicOperation {
+
+ @StageRef OwnerThreadHolder holder;
+
+ public void checkOnEachPublicOperation() {
+ holder.checkAccessingFromOwnerThread();
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/KeyBytesInterop.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/KeyBytesInterop.java
new file mode 100644
index 000000000..60653eef4
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/KeyBytesInterop.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.hash;
+
+import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
+import net.openhft.chronicle.hash.serialization.BytesReader;
+import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public class KeyBytesInterop> {
+ @StageRef
+ VanillaChronicleHashHolder hh;
+ @StageRef ThreadLocalCopiesHolder ch;
+
+ public final BytesReader keyReader =
+ hh.h().keyReaderProvider.get(ch.copies, hh.h().originalKeyReader);
+ public final KI keyInterop = hh.h().keyInteropProvider.get(ch.copies, hh.h().originalKeyInterop);
+
+ public MKI keyMetaInterop(K key) {
+ return hh.h().metaKeyInteropProvider.get(
+ ch.copies, hh.h().originalMetaKeyInterop, keyInterop, key);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/LogHolder.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/LogHolder.java
new file mode 100644
index 000000000..168bf073f
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/LogHolder.java
@@ -0,0 +1,10 @@
+package net.openhft.chronicle.hash.impl.stage.hash;
+
+import net.openhft.sg.Staged;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Staged
+public class LogHolder {
+ public static final Logger LOG = LoggerFactory.getLogger(LogHolder.class);
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/OwnerThreadHolder.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/OwnerThreadHolder.java
new file mode 100644
index 000000000..f95aa5deb
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/OwnerThreadHolder.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.hash;
+
+import net.openhft.sg.Staged;
+
+import java.util.ConcurrentModificationException;
+
+@Staged
+public class OwnerThreadHolder {
+
+ final Thread owner = Thread.currentThread();
+
+ public void checkAccessingFromOwnerThread() {
+ if (owner != Thread.currentThread()) {
+ throw new ConcurrentModificationException(
+ "Context shouldn't be accessed from multiple threads");
+ }
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalCopiesHolder.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalCopiesHolder.java
new file mode 100644
index 000000000..09151740e
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalCopiesHolder.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.hash;
+
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
+import net.openhft.sg.Staged;
+
+@Staged
+public class ThreadLocalCopiesHolder {
+ public final ThreadLocalCopies copies = ThreadLocalCopies.get();
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/input/HashInputBytes.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/input/HashInputBytes.java
new file mode 100644
index 000000000..e6f295d52
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/input/HashInputBytes.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.input;
+
+import net.openhft.chronicle.hash.impl.JavaLangBytesReusableBytesStore;
+import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
+import net.openhft.lang.io.Bytes;
+import net.openhft.sg.Stage;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public class HashInputBytes {
+
+ @StageRef VanillaChronicleHashHolder, ?, ?> hh;
+
+ public Bytes inputBytes = null;
+ public final JavaLangBytesReusableBytesStore inputStore = new JavaLangBytesReusableBytesStore();
+
+ public void initInputBytes(Bytes inputBytes) {
+ this.inputBytes = inputBytes;
+ inputStore.setBytes(inputBytes);
+ }
+
+ @Stage("InputKeyOffsets") public long inputKeySize = -1;
+ @Stage("InputKeyOffsets") public long inputKeyOffset;
+
+ private void initInputKeyOffsets() {
+ inputKeySize = hh.h().keySizeMarshaller.readSize(inputBytes);
+ inputKeyOffset = inputBytes.position();
+ }
+}
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
new file mode 100644
index 000000000..5d902c17e
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
@@ -0,0 +1,112 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.iter;
+
+import net.openhft.chronicle.hash.HashEntry;
+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;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+import java.util.function.Predicate;
+
+@Staged
+public abstract class HashSegmentIteration> implements HashEntry {
+
+ @StageRef public SegmentStages s;
+ @StageRef HashEntryStages e;
+ @StageRef protected HashLookup hashLookup;
+ @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
+ @StageRef protected HashLookupPos hlp;
+
+ public boolean entryIsPresent() {
+ return true;
+ }
+
+ public boolean entryRemovedOnThisIteration = false;
+
+ abstract boolean entryRemovedOnThisIterationInit();
+
+ protected void initEntryRemovedOnThisIteration(boolean entryRemovedOnThisIteration) {
+ this.entryRemovedOnThisIteration = entryRemovedOnThisIteration;
+ }
+
+ public boolean forEachRemoving(Predicate super E> action) {
+ s.innerUpdateLock.lock();
+ try {
+ long size = s.size();
+ if (size == 0)
+ return true;
+ boolean interrupted = false;
+ long startPos = 0L;
+ while (!hashLookup.empty(hashLookup.readEntry(startPos))) {
+ startPos = hashLookup.step(startPos);
+ }
+ hlp.initHashLookupPos(startPos);
+ do {
+ hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos));
+ long entry = hashLookup.readEntry(hlp.hashLookupPos);
+ if (!hashLookup.empty(entry)) {
+ e.readExistingEntry(hashLookup.value(entry));
+ if (entryIsPresent()) {
+ initEntryRemovedOnThisIteration(false);
+ if (!action.test((E) this)) {
+ interrupted = true;
+ break;
+ } else {
+ if (--size == 0)
+ break;
+ }
+
+ }
+ }
+ } while (hlp.hashLookupPos != startPos);
+ return !interrupted;
+ } finally {
+ s.innerReadLock.unlock();
+ initEntryRemovedOnThisIteration(false);
+ }
+ }
+
+ public void checkEntryNotRemovedOnThisIteration() {
+ if (entryRemovedOnThisIterationInit())
+ throw new IllegalStateException("Entry was already removed on this iteration");
+ }
+
+ @Override
+ public void doRemove() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ initEntryRemovedOnThisIteration(true);
+ s.innerWriteLock.lock();
+ try {
+ // this condition mean -- some other entry taken place of the removed one
+ if (hashLookup.remove(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));
+ }
+ e.innerRemoveEntryExceptHashLookupUpdate();
+ } finally {
+ s.innerWriteLock.unlock();
+ }
+ }
+}
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
new file mode 100644
index 000000000..b4d981729
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashLookupSearch.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.query;
+
+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;
+
+@Staged
+public abstract class HashLookupSearch {
+
+ @StageRef SegmentStages s;
+ @StageRef HashEntryStages> e;
+ @StageRef HashLookup hl;
+ @StageRef HashQuery op;
+ @StageRef VanillaChronicleHashHolder, ?, ?> hh;
+ @StageRef HashLookupPos hlp;
+
+ @Stage("SearchKey") long searchKey = UNSET_KEY;
+ @Stage("SearchKey") long searchStartPos;
+
+ void initSearchKey() {
+ searchKey = hl.maskUnsetKey(hh.h().hashSplitting.segmentHash(op.hashOfKey));
+ searchStartPos = hl.hlPos(searchKey);
+ }
+
+ public long nextPos() {
+ long pos = hlp.hashLookupPos;
+ while (true) {
+ long entry = hl.readEntry(pos);
+ if (hl.empty(entry)) {
+ hlp.setHashLookupPos(pos);
+ return -1L;
+ }
+ pos = hl.step(pos);
+ if (pos == searchStartPos)
+ break;
+ if (hl.key(entry) == searchKey) {
+ hlp.setHashLookupPos(pos);
+ return hl.value(entry);
+ }
+ }
+ throw new IllegalStateException("MultiMap is full, that most likely means you " +
+ "misconfigured entrySize/chunkSize, and entries tend to take less chunks than " +
+ "expected");
+ }
+
+ public void found() {
+ hlp.setHashLookupPos(hl.stepBack(hlp.hashLookupPos));
+ }
+
+ public void remove() {
+ hlp.setHashLookupPos(hl.remove(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);
+ }
+
+ public boolean checkSlotContainsExpectedKeyAndValue(long value) {
+ long entry = hl.readEntry(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
new file mode 100644
index 000000000..c03735ae5
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java
@@ -0,0 +1,155 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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;
+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.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;
+
+@Staged
+public abstract class HashQuery implements HashEntry {
+
+ @StageRef public VanillaChronicleHashHolder hh;
+ @StageRef public SegmentStages s;
+ @StageRef public HashEntryStages entry;
+ @StageRef public HashLookupSearch hashLookupSearch;
+ @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
+ @StageRef public HashLookupPos hlp;
+
+ @StageRef public InputKeyInstanceData inputKeyInstanceValue;
+ @StageRef public InputKeyBytesData inputKeyBytesValue;
+
+ public void dropSearchIfNestedContextsAndPresentHashLookupSlotCheckFailed() {
+ if (s.locksInit()) {
+ if (s.concurrentSameThreadContexts &&
+ s.rootContextOnThisSegment.latestSameThreadSegmentModCount !=
+ s.contextModCount) {
+ if (keySearchInit()) {
+ if (searchState == PRESENT) {
+ if (!hashLookupSearch.checkSlotContainsExpectedKeyAndValue(entry.pos))
+ hlp.closeHashLookupPos();
+ }
+ }
+ }
+ }
+ }
+
+ public Data inputKey = null;
+
+ public void initInputKey(Data inputKey) {
+ this.inputKey = inputKey;
+ }
+
+ public Data queriedKey() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ return 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();
+ }
+
+ public boolean entryPresent() {
+ return searchStatePresent();
+ }
+
+ @Override
+ public void doRemove() {
+ checkOnEachPublicOperation.checkOnEachPublicOperation();
+ s.innerUpdateLock.lock();
+ if (searchStatePresent()) {
+ // TODO optimize: if shift-deletion is trivial, updateLock.lock()
+ s.innerWriteLock.lock();
+ hashLookupSearch.remove();
+ entry.innerRemoveEntryExceptHashLookupUpdate();
+ 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/QueryHashLookupPos.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java
new file mode 100644
index 000000000..ae2d54ff2
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryHashLookupPos.java
@@ -0,0 +1,18 @@
+package net.openhft.chronicle.hash.impl.stage.query;
+
+import net.openhft.chronicle.hash.impl.stage.entry.HashLookupPos;
+import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public abstract class QueryHashLookupPos extends HashLookupPos {
+
+ @StageRef SegmentStages s;
+ @StageRef HashLookupSearch hashLookupSearch;
+
+ public void initHashLookupPos() {
+ s.innerReadLock.lock();
+ this.hashLookupPos = hashLookupSearch.searchStartPos;
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QuerySegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QuerySegmentStages.java
new file mode 100644
index 000000000..808e6a6d6
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QuerySegmentStages.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.query;
+
+import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public abstract class QuerySegmentStages extends SegmentStages {
+
+ @StageRef HashQuery q;
+
+ void initTheSegmentIndex() {
+ segmentIndex = hh.h().hashSplitting.segmentIndex(q.hashOfKey);
+ }
+}
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
new file mode 100644
index 000000000..c04a50e1e
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.impl.stage.query;
+
+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;
+
+@Staged
+public class SearchAllocatedChunks extends AllocatedChunks {
+
+ @StageRef HashQuery q;
+
+ @Override
+ public void incrementSegmentEntriesIfNeeded() {
+ if (q.searchState != PRESENT) {
+ // update the size before the store fence
+ s.entries(s.entries() + 1L);
+ }
+ }
+
+ public void initEntryAndKey(long entrySize) {
+ initAllocatedChunks(hh.h().inChunks(entrySize));
+ entry.writeNewEntry(s.alloc(allocatedChunks), q.inputKey);
+ incrementSegmentEntriesIfNeeded();
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
index a475889e1..6787259e9 100644
--- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
@@ -17,9 +17,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.function.SerializableFunction;
-import net.openhft.chronicle.hash.impl.hashlookup.HashLookupIteration;
import net.openhft.chronicle.hash.impl.util.CharSequences;
-import net.openhft.chronicle.map.impl.CompiledMapIterationContext;
import net.openhft.chronicle.map.impl.IterationContextInterface;
import org.jetbrains.annotations.NotNull;
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index 3f1e76416..bb2b8ea79 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -23,7 +23,8 @@ import net.openhft.chronicle.hash.replication.AbstractReplication;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
-import net.openhft.chronicle.map.impl.*;
+import net.openhft.chronicle.map.impl.CompiledReplicatedMapIterationContext;
+import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.map.replication.MapReplicableEntry;
import net.openhft.lang.Maths;
@@ -522,7 +523,8 @@ public class ReplicatedChronicleMap 0L;
+ // Bizarrely the next line line cause NPE in JDT compiler
+ //assert re.originTimestamp() > 0L;
if (debugEnabled) {
- LOG.debug("Bootstrap decision: bs ts: {}, entry ts: {}, entry id: {}, " +
- "local id: {}", fromTimeStamp, re.originTimestamp(),
+ LOG.debug("Bootstrap decision: bs ts: {}, entry ts: {}, " +
+ "entry id: {}, local id: {}",
+ fromTimeStamp, re.originTimestamp(),
re.originIdentifier(), localIdentifier);
}
if (re.originTimestamp() >= fromTimeStamp &&
diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
index 44944e328..5b790876c 100644
--- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
@@ -24,11 +24,9 @@ import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
import net.openhft.chronicle.hash.serialization.BytesReader;
-import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.ReaderWithSize;
import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext;
-import net.openhft.chronicle.map.impl.QueryContextInterface;
import net.openhft.lang.io.AbstractBytes;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
index 92c3f35dd..d66d06b07 100755
--- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
@@ -18,8 +18,8 @@
package net.openhft.chronicle.map;
-import net.openhft.chronicle.hash.KeyContext;
import net.openhft.chronicle.hash.Data;
+import net.openhft.chronicle.hash.KeyContext;
import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.SizeMarshaller;
@@ -27,6 +27,8 @@ import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.MetaProvider;
import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.map.impl.*;
+import net.openhft.chronicle.map.impl.CompiledMapIterationContext;
+import net.openhft.chronicle.map.impl.CompiledMapQueryContext;
import net.openhft.chronicle.map.impl.ret.InstanceReturnValue;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.model.DataValueClasses;
@@ -265,11 +267,13 @@ public class VanillaChronicleMap, V, VI, MVI extends MetaBytesInterop, R> implements AutoCloseable , HashEntry , InterProcessReadWriteUpdateLock , MapContext , MapEntry , IterationContextInterface , VanillaChronicleMapHolder {
- public void close() {
- this.closeTheSegmentIndex();
- this.wrappedValueInstanceValue.closeValue();
- this.closeEntryRemovedOnThisIteration();
- this.wrappedValueInstanceValue.closeNext();
- this.closeUsed();
- this.closeKeySize();
- this.closePos();
- this.closeAllocatedChunks();
- this.closeHashLookupPos();
- this.closeKeyOffset();
- this.closeOwnerThreadHolderCheckAccessingFromOwnerThreadDependants();
- this.closeVanillaChronicleMapHolderImplContextAtIndexInChainDependants();
- this.closeValueBytesInteropValueMetaInteropDependants();
- this.closeMapSegmentIterationCheckEntryNotRemovedOnThisIterationDependants();
- }
-
- public void incrementModCountGuarded() {
- if (!(this.locksInit()))
- this.initLocks();
-
- incrementModCount();
- }
-
- public void setHashLookupPosGuarded(long hashLookupPos) {
- assert this.hashLookupPosInit() : "HashLookupPos should be init";
- setHashLookupPos(hashLookupPos);
- }
-
- public void setLocalLockStateGuarded(LocalLockState newState) {
- if (!(this.locksInit()))
- this.initLocks();
-
- setLocalLockState(newState);
- }
-
- public CompiledMapIterationContext(VanillaChronicleMap m) {
- contextChain = new ArrayList();
- contextChain.add(this);
- indexInContextChain = 0;
- this.m = m;
- this.copies = ThreadLocalCopies.get();
- this.valueInterop = this.m().valueInteropProvider.get(this.copies, this.m().originalValueInterop);
- this.valueReader = this.m().valueReaderProvider.get(this.copies, this.m().originalValueReader);
- this.keyInterop = this.h().keyInteropProvider.get(this.copies, this.h().originalKeyInterop);
- this.keyReader = this.h().keyReaderProvider.get(this.copies, this.h().originalKeyReader);
- this.entryKey = new EntryKeyBytesData();
- this.innerUpdateLock = new UpdateLock();
- this.innerReadLock = new ReadLock();
- this.entryValue = new EntryValueBytesData();
- this.innerWriteLock = new WriteLock();
- this.entryBytes = this.h().ms.bytes();
- this.entryBS = new NativeBytesStore(entryBytes.address() , entryBytes.capacity() , null , false);
- this.owner = Thread.currentThread();
- this.deprecatedMapKeyContextOnIteration = new DeprecatedMapKeyContextOnIteration();
- this.wrappedValueInstanceValue = new WrappedValueInstanceData();
- }
-
- public CompiledMapIterationContext(CompiledMapIterationContext c) {
- contextChain = c.contextChain;
- indexInContextChain = contextChain.size();
- contextChain.add(this);
- this.m = ((VanillaChronicleMap)(c.m));
- this.copies = ThreadLocalCopies.get();
- this.valueInterop = this.m().valueInteropProvider.get(this.copies, this.m().originalValueInterop);
- this.valueReader = this.m().valueReaderProvider.get(this.copies, this.m().originalValueReader);
- this.keyInterop = this.h().keyInteropProvider.get(this.copies, this.h().originalKeyInterop);
- this.keyReader = this.h().keyReaderProvider.get(this.copies, this.h().originalKeyReader);
- this.entryKey = new EntryKeyBytesData();
- this.innerUpdateLock = new UpdateLock();
- this.innerReadLock = new ReadLock();
- this.entryValue = new EntryValueBytesData();
- this.innerWriteLock = new WriteLock();
- this.entryBytes = this.h().ms.bytes();
- this.entryBS = new NativeBytesStore