diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/Alloc.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/Alloc.java
new file mode 100644
index 000000000..3ff9bee52
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/Alloc.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright (C) 2015 higherfrequencytrading.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program. If not, see .
+ */
+
+package net.openhft.chronicle.hash.impl.stage.entry;
+
+public interface Alloc {
+
+ /**
+ * Allocates a position in a segment tier.
+ *
+ * @param chunks chunks to allocate
+ * @return the position, if allocation is successful, or -1
+ */
+ long alloc(int chunks);
+}
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
index 77988a950..268946336 100644
--- 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
@@ -26,6 +26,7 @@ public class AllocatedChunks {
@StageRef public VanillaChronicleHashHolder hh;
@StageRef public SegmentStages s;
@StageRef public HashEntryStages entry;
+ @StageRef public Alloc alloc;
public int allocatedChunks = 0;
@@ -50,8 +51,8 @@ public class AllocatedChunks {
long oldKeySizeAddr = oldSegmentTierBaseAddr + entry.keySizeOffset;
long oldKeyAddr = oldSegmentTierBaseAddr + entry.keyOffset;
int tierBeforeAllocation = s.segmentTier;
- long pos = s.alloc(allocatedChunks);
+ long pos = alloc.alloc(allocatedChunks);
entry.copyExistingEntry(pos, bytesToCopy, oldKeyAddr, oldKeySizeAddr);
- return s.segmentTier > tierBeforeAllocation;
+ return s.segmentTier != tierBeforeAllocation;
}
}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
index 53d727d46..ca4320bcb 100644
--- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
@@ -524,6 +524,18 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
throw new IllegalStateException("first tier doesn't have previous");
initSegmentTier(segmentTier - 1, prevTierIndex());
}
+
+ public void goToLastTier() {
+ while (hasNextTier()) {
+ nextTier();
+ }
+ }
+
+ public void goToFirstTier() {
+ while (segmentTier != 0) {
+ prevTier();
+ }
+ }
@Stage("Segment") public final PublicMultiStoreBytes segmentBytes = new PublicMultiStoreBytes();
@Stage("Segment") public final PointerBytesStore segmentBS = new PointerBytesStore();
@@ -589,17 +601,6 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
return ret;
}
- public long alloc(int chunks) {
- while (true) {
- long ret = allocReturnCode(chunks);
- if (ret >= 0) {
- return ret;
- } else {
- nextTier();
- }
- }
- }
-
public void free(long fromPos, int chunks) {
freeList.clearRange(fromPos, fromPos + chunks);
if (fromPos < nextPosToSearchFrom())
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
index cd35cc651..55552b6eb 100644
--- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
@@ -62,12 +62,6 @@ public abstract class HashSegmentIteration>
abstract void closeHashLookupEntry();
- private void goToLastTier() {
- while (s.hasNextTier()) {
- s.nextTier();
- }
- }
-
@Override
public boolean forEachSegmentEntryWhile(Predicate action) {
s.innerUpdateLock.lock();
@@ -75,7 +69,7 @@ public abstract class HashSegmentIteration>
long size = s.size();
if (size == 0)
return true;
- goToLastTier();
+ s.goToLastTier();
while (true) {
int currentTier = s.segmentTier;
long currentTierBaseAddr = s.segmentBaseAddr;
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationAlloc.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationAlloc.java
new file mode 100644
index 000000000..97a73098f
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationAlloc.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright (C) 2015 higherfrequencytrading.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program. If not, see .
+ */
+
+package net.openhft.chronicle.hash.impl.stage.iter;
+
+import net.openhft.chronicle.hash.impl.stage.entry.Alloc;
+import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public class IterationAlloc implements Alloc {
+
+ @StageRef public SegmentStages s;
+
+ /**
+ * Move only to next tiers, to avoid double visiting of relocated entries during iteration
+ */
+ @Override
+ public long alloc(int chunks) {
+ while (true) {
+ long ret = s.allocReturnCode(chunks);
+ if (ret >= 0) {
+ return ret;
+ } else {
+ s.nextTier();
+ }
+ }
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryAlloc.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryAlloc.java
new file mode 100644
index 000000000..687082c19
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryAlloc.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2015 higherfrequencytrading.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program. If not, see .
+ */
+
+package net.openhft.chronicle.hash.impl.stage.query;
+
+import net.openhft.chronicle.hash.impl.stage.entry.Alloc;
+import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+@Staged
+public class QueryAlloc implements Alloc {
+
+ @StageRef public SegmentStages s;
+
+ @Override
+ public long alloc(int chunks) {
+ long ret = s.allocReturnCode(chunks);
+ if (ret >= 0)
+ return ret;
+ int alreadyAttemptedTier = s.segmentTier;
+ s.goToFirstTier();
+ while (true) {
+ if (s.segmentTier != alreadyAttemptedTier) {
+ ret = s.allocReturnCode(chunks);
+ if (ret >= 0)
+ return ret;
+ }
+ s.nextTier();
+ }
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java
index d15b47938..24ad23d91 100644
--- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java
@@ -44,8 +44,8 @@ public class SearchAllocatedChunks extends AllocatedChunks {
// clears out searchState, and it performs the search again, but in inconsistent state
incrementSegmentEntriesIfNeeded();
int tierBeforeAllocation = s.segmentTier;
- long pos = s.alloc(allocatedChunks);
+ long pos = alloc.alloc(allocatedChunks);
entry.writeNewEntry(pos, ks.inputKey);
- return s.segmentTier > tierBeforeAllocation;
+ return s.segmentTier != tierBeforeAllocation;
}
}
diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java
index 2b231e42c..dd03d40f1 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java
@@ -22,6 +22,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
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.iter.IterationAlloc;
import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode;
import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages;
import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
@@ -61,6 +62,8 @@ import net.openhft.sg.Staged;
MapEntryStages.class,
ValueBytesInterop.class,
MapEntryOperationsDelegation.class,
+
+ IterationAlloc.class,
}, nested = {
ReadLock.class,
UpdateLock.class,
diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
index 8e2e62e99..896f9ff32 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
@@ -70,6 +70,8 @@ import net.openhft.sg.Staged;
HashInputBytes.class,
MapInputBytesValues.class,
+
+ QueryAlloc.class,
},
nested = {
ReadLock.class,
diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java
index 6dbc88cee..5f53b05a0 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java
@@ -22,6 +22,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
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.iter.IterationAlloc;
import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode;
import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages;
import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
@@ -65,6 +66,8 @@ import net.openhft.sg.Staged;
ReplicationUpdate.class,
DefaultValue.class,
+
+ IterationAlloc.class,
}, nested = {
ReadLock.class,
UpdateLock.class,
diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java
index 6e1fbd54f..e728b8d65 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java
@@ -25,10 +25,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.KeySearch;
-import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupPos;
-import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupSearch;
-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;
@@ -38,6 +35,7 @@ import net.openhft.chronicle.map.impl.stage.input.MapInputBytesValues;
import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput;
import net.openhft.chronicle.map.impl.stage.map.*;
import net.openhft.chronicle.map.impl.stage.query.*;
+import net.openhft.chronicle.map.impl.stage.replication.ReplicatedQueryAlloc;
import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate;
import net.openhft.chronicle.map.impl.stage.ret.BytesReturnValue;
import net.openhft.chronicle.map.impl.stage.ret.DefaultReturnValue;
@@ -56,7 +54,7 @@ import net.openhft.sg.Staged;
ReplicatedChronicleMapHolderImpl.class,
KeyBytesInterop.class,
- ReplicatedQuerySegmentStages.class,
+ QuerySegmentStages.class,
KeySearch.class,
InputKeyHashCode.class,
QueryHashLookupSearch.class,
@@ -78,6 +76,8 @@ import net.openhft.sg.Staged;
HashInputBytes.class,
MapInputBytesValues.class,
+
+ ReplicatedQueryAlloc.class,
},
nested = {
ReadLock.class,
diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java
deleted file mode 100644
index 26a2b82d6..000000000
--- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Copyright (C) 2015 higherfrequencytrading.com
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Lesser General Public License as published by
- * the Free Software Foundation, either version 3 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public License
- * along with this program. If not, see .
- */
-
-package net.openhft.chronicle.map.impl.stage.query;
-
-import net.openhft.chronicle.hash.ReplicatedHashSegmentContext;
-import net.openhft.chronicle.hash.impl.stage.query.QuerySegmentStages;
-import net.openhft.chronicle.hash.replication.ReplicableEntry;
-import net.openhft.chronicle.map.MapAbsentEntry;
-import net.openhft.chronicle.map.MapSegmentContext;
-import net.openhft.chronicle.map.ReplicatedChronicleMap;
-import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder;
-import net.openhft.sg.StageRef;
-import net.openhft.sg.Staged;
-
-import java.util.function.Consumer;
-
-@Staged
-public abstract class ReplicatedQuerySegmentStages extends QuerySegmentStages {
-
- @StageRef
- ReplicatedChronicleMapHolder mh;
-
- final Consumer cleanupAction = e -> {
- ReplicatedChronicleMap map = mh.m();
- if (e instanceof MapAbsentEntry) {
- long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp();
- if (deleteTimeout > map.cleanupTimeout && !e.isChanged())
- e.doRemoveCompletely();
- }
- };
-
- @Override
- public long alloc(int chunks) {
- while (true) {
- long ret = allocReturnCode(chunks);
- if (ret >= 0) {
- return ret;
- } else {
- ReplicatedChronicleMap map = mh.m();
- try (MapSegmentContext sc = map.segmentContext(segmentIndex)) {
- ((ReplicatedHashSegmentContext) sc)
- .forEachSegmentReplicableEntry(cleanupAction);
- }
- ret = allocReturnCode(chunks);
- if (ret >= 0) {
- return ret;
- } else {
- nextTier();
- }
- }
- }
- }
-}
diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicatedQueryAlloc.java b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicatedQueryAlloc.java
new file mode 100644
index 000000000..73ae1bea5
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicatedQueryAlloc.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright (C) 2015 higherfrequencytrading.com
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program. If not, see .
+ */
+
+package net.openhft.chronicle.map.impl.stage.replication;
+
+import net.openhft.chronicle.hash.ReplicatedHashSegmentContext;
+import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
+import net.openhft.chronicle.hash.impl.stage.query.QueryAlloc;
+import net.openhft.chronicle.hash.replication.ReplicableEntry;
+import net.openhft.chronicle.map.MapAbsentEntry;
+import net.openhft.chronicle.map.MapSegmentContext;
+import net.openhft.chronicle.map.ReplicatedChronicleMap;
+import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder;
+import net.openhft.sg.StageRef;
+import net.openhft.sg.Staged;
+
+import java.util.function.Consumer;
+
+@Staged
+public class ReplicatedQueryAlloc extends QueryAlloc {
+
+ @StageRef ReplicatedChronicleMapHolder mh;
+ @StageRef SegmentStages s;
+
+ final CleanupAction cleanupAction = new CleanupAction();
+
+ /**
+ * Returns {@code true} if at least one old deleted entry was removed.
+ */
+ public boolean forcedOldDeletedEntriesCleanup() {
+ ReplicatedChronicleMap map = mh.m();
+ try (MapSegmentContext sc = map.segmentContext(s.segmentIndex)) {
+ cleanupAction.removedCompletely = 0;
+ ((ReplicatedHashSegmentContext) sc)
+ .forEachSegmentReplicableEntry(cleanupAction);
+ return cleanupAction.removedCompletely > 0;
+ }
+ }
+
+ private class CleanupAction implements Consumer {
+ int removedCompletely;
+
+ @Override
+ public void accept(ReplicableEntry e) {
+ ReplicatedChronicleMap map = mh.m();
+ if (e instanceof MapAbsentEntry) {
+ long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp();
+ map.timeProvider.systemTimeIntervalBetween(e.originTimestamp(),
+ map.timeProvider.currentTime(), map.cleanupTimeoutUnit);
+ if (deleteTimeout > map.cleanupTimeout && !e.isChanged()) {
+ e.doRemoveCompletely();
+ removedCompletely++;
+ }
+ }
+ }
+ }
+
+ @Override
+ public long alloc(int chunks) {
+ long ret = s.allocReturnCode(chunks);
+ if (ret >= 0)
+ return ret;
+ int alreadyAttemptedTier;
+ if (!forcedOldDeletedEntriesCleanup()) {
+ alreadyAttemptedTier = s.segmentTier;
+ } else {
+ // If forced old deleted entries cleanup deleted something, don't skip the first queried
+ // tier, because it could have new empty chunks now. segment tier is >= 0, so set to -1
+ // to make the condition with alreadyAttemptedTier always false
+ alreadyAttemptedTier = -1;
+ }
+ s.goToFirstTier();
+ while (true) {
+ if (s.segmentTier != alreadyAttemptedTier) {
+ ret = s.allocReturnCode(chunks);
+ if (ret >= 0)
+ return ret;
+ }
+ s.nextTier();
+ }
+ }
+}