diff --git a/pom.xml b/pom.xml index 726c865bd..1780500b4 100755 --- a/pom.xml +++ b/pom.xml @@ -148,7 +148,7 @@ net.openhft stage-compiler - 1.2 + 1.3 provided diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java index dda304ece..49d4fccfc 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -16,6 +16,8 @@ package net.openhft.chronicle.hash.impl; +import net.openhft.chronicle.algo.bytes.Access; +import net.openhft.chronicle.algo.locks.*; import net.openhft.chronicle.bytes.NativeBytesStore; import net.openhft.chronicle.core.Maths; import net.openhft.chronicle.hash.*; @@ -39,6 +41,7 @@ import java.io.Serializable; import java.nio.channels.FileChannel; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.TimeUnit; import static java.lang.Long.numberOfTrailingZeros; import static java.lang.Math.max; @@ -452,6 +455,18 @@ public abstract class VanillaChronicleHash tierDataLockTryAcquireOperation = + TryAcquireOperations.lock(); + static final + AcquisitionStrategy tierDataLockAcquisitionStrategy = + AcquisitionStrategies.spinLoopOrFail(2, TimeUnit.SECONDS); + private static final long TIER_DATA_LOCK_OFFSET = 0L; private static final long EXTRA_ALLOCATED_TIER_BULK_COUNT_OFFSET = TIER_DATA_LOCK_OFFSET + 8L; private static final long FIRST_FREE_TIER_INDEX_OFFSET = @@ -463,13 +478,14 @@ public abstract class VanillaChronicleHash> log2NumberOfTiersInBulk; + if (bulkIndex >= tierBulkOffsets.size()) + mapTiers(bulkIndex); + return tierBulkOffsets.get((int) bulkIndex).offset + tierBulkInnerOffsetToTiers + + (extraTierIndex & (numberOfTiersInBulk - 1)) * segmentSize; } private TierBulkData tierBulkData(long tierIndex) { diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/LocksInterface.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/LocksInterface.java new file mode 100644 index 000000000..2bdb7d8f7 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/LocksInterface.java @@ -0,0 +1,52 @@ +/* + * 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 LocksInterface { + + boolean segmentHeaderInit(); + + long segmentHeaderAddress(); + + boolean locksInit(); + + LocksInterface rootContextOnThisSegment(); + + void setConcurrentSameThreadContexts(boolean concurrentSameThreadContexts); + + int latestSameThreadSegmentModCount(); + + int changeAndGetLatestSameThreadSegmentModCount(int change); + + int totalReadLockCount(); + + int changeAndGetTotalReadLockCount(int change); + + int totalUpdateLockCount(); + + int changeAndGetTotalUpdateLockCount(int change); + + int totalWriteLockCount(); + + int changeAndGetTotalWriteLockCount(int change); + + LocksInterface nextNode(); + + void setNextNode(LocksInterface nextNode); + + String debugLocksState(); +} 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 348bb5330..ee2fe9a7f 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 @@ -25,11 +25,11 @@ import net.openhft.chronicle.hash.SegmentLock; import net.openhft.chronicle.hash.impl.*; import net.openhft.chronicle.hash.impl.stage.hash.Chaining; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; +import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.locks.InterProcessLock; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.lang.collection.DirectBitSet; -import net.openhft.lang.io.Bytes; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -42,7 +42,7 @@ import static net.openhft.chronicle.algo.MemoryUnit.LONGS; import static net.openhft.chronicle.hash.impl.LocalLockState.UNLOCKED; @Staged -public abstract class SegmentStages implements SegmentLock { +public abstract class SegmentStages implements SegmentLock, LocksInterface { @StageRef Chaining chaining; @StageRef public VanillaChronicleHashHolder hh; @@ -54,11 +54,11 @@ public abstract class SegmentStages implements SegmentLock { this.segmentIndex = segmentIndex; } + public abstract boolean segmentIndexInit(); + @Stage("SegmentHeader") long segmentHeaderAddress; @Stage("SegmentHeader") SegmentHeader segmentHeader = null; - abstract boolean segmentHeaderInit(); - private void initSegmentHeader() { segmentHeaderAddress = hh.h().segmentHeaderAddress(segmentIndex); segmentHeader = BigSegmentHeader.INSTANCE; @@ -101,7 +101,7 @@ public abstract class SegmentStages implements SegmentLock { } - @Stage("Locks") public SegmentStages rootContextOnThisSegment = null; + @Stage("Locks") public LocksInterface 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, @@ -112,16 +112,35 @@ public abstract class SegmentStages implements SegmentLock { */ @Stage("Locks") public boolean concurrentSameThreadContexts; + @Override + @Stage("Locks") + public void setConcurrentSameThreadContexts(boolean concurrentSameThreadContexts) { + this.concurrentSameThreadContexts = concurrentSameThreadContexts; + } + @Stage("Locks") public int latestSameThreadSegmentModCount; + + @Override + @Stage("Locks") + public int changeAndGetLatestSameThreadSegmentModCount(int change) { + return this.latestSameThreadSegmentModCount += change; + } + @Stage("Locks") public int contextModCount; @Stage("Locks") public void incrementModCount() { - contextModCount = ++rootContextOnThisSegment.latestSameThreadSegmentModCount; + contextModCount = rootContextOnThisSegment.changeAndGetLatestSameThreadSegmentModCount(1); } // chain - @Stage("Locks") SegmentStages nextNode; + @Stage("Locks") LocksInterface nextNode; + + @Override + @Stage("Locks") + public void setNextNode(LocksInterface nextNode) { + this.nextNode = nextNode; + } @Stage("Locks") LocalLockState localLockState; @Stage("Locks") int totalReadLockCount; @@ -130,47 +149,65 @@ public abstract class SegmentStages implements SegmentLock { @Stage("Locks") public boolean readZero() { - return rootContextOnThisSegment.totalReadLockCount == 0; + return rootContextOnThisSegment.totalReadLockCount() == 0; + } + + @Override + @Stage("Locks") + public int changeAndGetTotalReadLockCount(int change) { + return totalReadLockCount += change; } @Stage("Locks") public boolean updateZero() { - return rootContextOnThisSegment.totalUpdateLockCount == 0; + return rootContextOnThisSegment.totalUpdateLockCount() == 0; + } + + @Override + @Stage("Locks") + public int changeAndGetTotalUpdateLockCount(int change) { + return totalUpdateLockCount += change; } @Stage("Locks") public boolean writeZero() { - return rootContextOnThisSegment.totalWriteLockCount == 0; + return rootContextOnThisSegment.totalWriteLockCount() == 0; + } + + @Override + @Stage("Locks") + public int changeAndGetTotalWriteLockCount(int change) { + return totalWriteLockCount += change; } @Stage("Locks") public int decrementRead() { - return rootContextOnThisSegment.totalReadLockCount -= 1; + return rootContextOnThisSegment.changeAndGetTotalReadLockCount(-1); } @Stage("Locks") public int decrementUpdate() { - return rootContextOnThisSegment.totalUpdateLockCount -= 1; + return rootContextOnThisSegment.changeAndGetTotalUpdateLockCount(-1); } @Stage("Locks") public int decrementWrite() { - return rootContextOnThisSegment.totalWriteLockCount -= 1; + return rootContextOnThisSegment.changeAndGetTotalWriteLockCount(-1); } @Stage("Locks") public void incrementRead() { - rootContextOnThisSegment.totalReadLockCount += 1; + rootContextOnThisSegment.changeAndGetTotalReadLockCount(1); } @Stage("Locks") public void incrementUpdate() { - rootContextOnThisSegment.totalUpdateLockCount += 1; + rootContextOnThisSegment.changeAndGetTotalUpdateLockCount(1); } @Stage("Locks") public void incrementWrite() { - rootContextOnThisSegment.totalWriteLockCount += 1; + rootContextOnThisSegment.changeAndGetTotalWriteLockCount(1); } public abstract boolean locksInit(); @@ -179,11 +216,11 @@ public abstract class SegmentStages implements SegmentLock { localLockState = UNLOCKED; int indexOfThisContext = chaining.indexInContextChain; for (int i = indexOfThisContext - 1; i >= 0; i--) { - if (tryFindInitLocksOfThisSegment(this, i)) + if (tryFindInitLocksOfThisSegment(i)) return; } for (int i = indexOfThisContext + 1, size = chaining.contextChain.size(); i < size; i++) { - if (tryFindInitLocksOfThisSegment(this, i)) + if (tryFindInitLocksOfThisSegment(i)) return; } rootContextOnThisSegment = this; @@ -198,16 +235,16 @@ public abstract class SegmentStages implements SegmentLock { } @Stage("Locks") - boolean tryFindInitLocksOfThisSegment(Object thisContext, int index) { - SegmentStages c = chaining.contextAtIndexInChain(index); + boolean tryFindInitLocksOfThisSegment(int index) { + LocksInterface c = chaining.contextAtIndexInChain(index); if (c.segmentHeaderInit() && - c.segmentHeaderAddress == segmentHeaderAddress && + c.segmentHeaderAddress() == segmentHeaderAddress && c.locksInit()) { - SegmentStages root = c.rootContextOnThisSegment; + LocksInterface root = c.rootContextOnThisSegment(); this.rootContextOnThisSegment = root; - root.concurrentSameThreadContexts = true; + root.setConcurrentSameThreadContexts(true); this.concurrentSameThreadContexts = true; - this.contextModCount = root.latestSameThreadSegmentModCount; + this.contextModCount = root.latestSameThreadSegmentModCount(); linkToSegmentContextsChain(); return true; } else { @@ -279,32 +316,34 @@ public abstract class SegmentStages implements SegmentLock { @Stage("Locks") private void linkToSegmentContextsChain() { - SegmentStages innermostContextOnThisSegment = rootContextOnThisSegment; + LocksInterface innermostContextOnThisSegment = rootContextOnThisSegment; while (true) { - Data key = ((KeySearch) (Object) innermostContextOnThisSegment).inputKey; - if (Objects.equals(key, ((KeySearch) (Object) this).inputKey)) { - throw new IllegalStateException("Nested same-thread contexts cannot access " + - "the same key " + key); + if (innermostContextOnThisSegment instanceof KeySearch) { + Data key = ((KeySearch) innermostContextOnThisSegment).inputKey; + if (Objects.equals(key, ((KeySearch) (Object) this).inputKey)) { + throw new IllegalStateException("Nested same-thread contexts cannot access " + + "the same key " + key); + } } - if (innermostContextOnThisSegment.nextNode == null) + if (innermostContextOnThisSegment.nextNode() == null) break; - innermostContextOnThisSegment = innermostContextOnThisSegment.nextNode; + innermostContextOnThisSegment = innermostContextOnThisSegment.nextNode(); } - innermostContextOnThisSegment.nextNode = this; + innermostContextOnThisSegment.setNextNode(this); } @Stage("Locks") private void unlinkFromSegmentContextsChain() { - SegmentStages prevContext = rootContextOnThisSegment; + LocksInterface prevContext = rootContextOnThisSegment; while (true) { - assert prevContext.nextNode != null; - if (prevContext.nextNode == this) + assert prevContext.nextNode() != null; + if (prevContext.nextNode() == this) break; - prevContext = prevContext.nextNode; + prevContext = prevContext.nextNode(); } // i. e. structured unlocking verifyInnermostContext(); - prevContext.nextNode = null; + prevContext.setNextNode(null); } @Stage("Locks") @@ -336,6 +375,46 @@ public abstract class SegmentStages implements SegmentLock { localLockState = newState; } + @Stage("Locks") + public String debugContextsAndLocks() { + String message = ""; + message += "Contexts locked on this segment:\n"; + + for (LocksInterface cxt = rootContextOnThisSegment; cxt != null; cxt = cxt.nextNode()) { + message += cxt.debugLocksState() + "\n"; + } + message += "Current thread contexts:\n"; + for (int i = 0, size = chaining.contextChain.size(); i < size; i++) { + LocksInterface cxt = chaining.contextAtIndexInChain(i); + message += cxt.debugLocksState() + "\n"; + } + return message; + } + + @Override + public String debugLocksState() { + String s = this + ": "; + if (!chaining.usedInit()) { + s += "unused"; + return s; + } + s += "used, "; + if (!segmentIndexInit()) { + s += "segment uninitialized"; + return s; + } + s += "segment " + segmentIndex() + ", "; + if (!locksInit()) { + s += "locks uninitialized"; + return s; + } + s += "local state: " + localLockState + ", "; + s += "read lock count: " + rootContextOnThisSegment.totalReadLockCount() + ", "; + s += "update lock count: " + rootContextOnThisSegment.totalUpdateLockCount() + ", "; + s += "write lock count: " + rootContextOnThisSegment.totalWriteLockCount(); + return s; + } + @StageRef public ReadLock innerReadLock; @StageRef public UpdateLock innerUpdateLock; @StageRef public WriteLock innerWriteLock; @@ -421,9 +500,9 @@ public abstract class SegmentStages implements SegmentLock { if (nextTierIndex == 0) { nextTierIndex = h.allocateTier(segmentIndex, segmentTier + 1); nextTierIndex(nextTierIndex); - long currentTierBase = segmentBaseAddr; + long currentTierIndex = tierIndex; initSegmentTier(segmentTier + 1, nextTierIndex); - prevTierIndex(currentTierBase); + prevTierIndex(currentTierIndex); } else { initSegmentTier(segmentTier + 1, nextTierIndex); } @@ -456,12 +535,12 @@ public abstract class SegmentStages implements SegmentLock { PublicMultiStoreBytes segmentBytes = this.segmentBytes; segmentBytes.setBytesOffset(h.tierBytes(tierIndex), h.bytesOffset(tierIndex)); - long segmentBase = this.segmentBaseAddr; - segmentBS.set(segmentBase, h.segmentSize); + long segmentBaseAddr = this.segmentBaseAddr; + segmentBS.set(segmentBaseAddr, h.segmentSize); // 64 is a cache line with tier data long freeListOffset = h.segmentHashLookupOuterSize + 64; - freeList.setOffset(segmentBase + freeListOffset); + freeList.setOffset(segmentBaseAddr + freeListOffset); entrySpaceOffset = freeListOffset + h.segmentFreeListOuterSize + h.segmentEntrySpaceInnerOffset; 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 index f419993a6..39f8e79fd 100644 --- 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 @@ -16,6 +16,7 @@ package net.openhft.chronicle.hash.impl.stage.entry; +import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import net.openhft.chronicle.hash.locks.InterProcessLock; @@ -29,6 +30,7 @@ import static net.openhft.chronicle.hash.impl.LocalLockState.UPDATE_LOCKED; @Staged public class UpdateLock implements InterProcessLock { + @StageRef LogHolder logHolder; @StageRef SegmentStages s; @Override @@ -43,7 +45,12 @@ public class UpdateLock implements InterProcessLock { if (s.updateZero() && s.writeZero()) { if (!s.readZero()) throw forbiddenUpdateLockWhenOuterContextReadLocked(); - s.segmentHeader.updateLock(s.segmentHeaderAddress); + try { + s.segmentHeader.updateLock(s.segmentHeaderAddress); + } catch (RuntimeException e) { + logHolder.LOG.error(s.debugContextsAndLocks()); + throw e; + } } s.incrementUpdate(); 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 index 10c234c02..a1fc27963 100644 --- 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 @@ -20,11 +20,12 @@ import net.openhft.sg.Staged; import java.util.ArrayList; import java.util.List; +import java.util.function.Function; @Staged -public class Chaining { +public class Chaining implements ChainingInterface { - public final List contextChain; + public final List contextChain; public final int indexInContextChain; public Chaining() { @@ -33,12 +34,17 @@ public class Chaining { indexInContextChain = 0; } - public Chaining(Chaining root) { - contextChain = root.contextChain; + public Chaining(ChainingInterface root) { + contextChain = root.getContextChain(); indexInContextChain = contextChain.size(); contextChain.add(this); } + @Override + public List getContextChain() { + return contextChain; + } + public T contextAtIndexInChain(int index) { //noinspection unchecked return (T) contextChain.get(index); @@ -46,7 +52,8 @@ public class Chaining { boolean used; - + + @Override public boolean usedInit() { return used; } @@ -59,9 +66,11 @@ public class Chaining { used = false; } - public T getContext() { - for (Chaining context : contextChain) { - if (!context.usedInit()) { + @Override + public T getContext( + Class contextClass, Function createChaining) { + for (ChainingInterface context : contextChain) { + if (context.getClass() == contextClass && !context.usedInit()) { //noinspection unchecked return (T) context; } @@ -76,10 +85,6 @@ public class Chaining { "stack trace on https://github.com/OpenHFT/Chronicle-Map/issues"); } //noinspection unchecked - return (T) createChaining(); - } - - public Chaining createChaining() { - return new Chaining(this); + return (T) createChaining.apply(this); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java new file mode 100644 index 000000000..9219c53e5 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java @@ -0,0 +1,30 @@ +/* + * 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.hash; + +import java.util.List; +import java.util.function.Function; + +public interface ChainingInterface { + + List getContextChain(); + + boolean usedInit(); + + T getContext( + Class contextClass, Function createChaining); +} 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 65af1d8e9..cd35cc651 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 @@ -54,6 +54,8 @@ public abstract class HashSegmentIteration> public long hashLookupEntry = 0; + public abstract boolean hashLookupEntryInit(); + public void initHashLookupEntry(long entry) { hashLookupEntry = entry; } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java index f809b9cc7..cf609d680 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java @@ -40,7 +40,8 @@ public abstract class IterationSegmentStages extends SegmentStages { @Override public void nextTier() { super.nextTier(); - hls.initSearchKey(hh.h().hashLookup.key(it.hashLookupEntry)); + if (it.hashLookupEntryInit()) + hls.initSearchKey(hh.h().hashLookup.key(it.hashLookupEntry)); } public void initSegmentTier_WithBaseAddr(int tier, long tierBaseAddr) { diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java index ecd679073..b15985a3d 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java @@ -50,7 +50,7 @@ public abstract class HashQuery implements HashEntry { public void dropSearchIfNestedContextsAndPresentHashLookupSlotCheckFailed() { if (s.locksInit()) { if (s.concurrentSameThreadContexts && - s.rootContextOnThisSegment.latestSameThreadSegmentModCount != + s.rootContextOnThisSegment.latestSameThreadSegmentModCount() != s.contextModCount) { if (ks.keySearchInit()) { if (ks.searchState == PRESENT) { diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index c2a0ff8d3..d01974174 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -21,6 +21,7 @@ import net.openhft.chronicle.algo.bitset.ConcurrentFlatBitSetFrame; import net.openhft.chronicle.algo.bitset.SingleThreadedFlatBitSetFrame; import net.openhft.chronicle.hash.impl.TierData; import net.openhft.chronicle.hash.impl.VanillaChronicleHash; +import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.hash.replication.AbstractReplication; import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.hash.replication.TimeProvider; @@ -139,30 +140,6 @@ public class ReplicatedChronicleMap>() { - @Override - protected CompiledReplicatedMapQueryContext - initialValue() { - return new CompiledReplicatedMapQueryContext<>(ReplicatedChronicleMap.this); - } - }; - } - - @Override - void initIterationContext() { - iterCxt = new ThreadLocal< - CompiledReplicatedMapIterationContext>() { - @Override - protected CompiledReplicatedMapIterationContext - initialValue() { - return new CompiledReplicatedMapIterationContext<>(ReplicatedChronicleMap.this); - } - }; - } - private int assignedModIterBitSetSizeInBytes() { return (int) CACHE_LINES.align(BYTES.alignAndConvert(127 + RESERVED_MOD_ITER, BITS), BYTES); } @@ -471,14 +448,22 @@ public class ReplicatedChronicleMap q() { - //noinspection unchecked - return (CompiledReplicatedMapQueryContext) queryCxt.get(); + private ChainingInterface q() { + ChainingInterface queryContext; + queryContext = cxt.get(); + if (queryContext == null) { + queryContext = new CompiledReplicatedMapQueryContext<>(ReplicatedChronicleMap.this); + cxt.set(queryContext); + } + return queryContext; } @Override public CompiledReplicatedMapQueryContext mapContext() { - CompiledReplicatedMapQueryContext q = q().getContext(); + CompiledReplicatedMapQueryContext q = + q().getContext(CompiledReplicatedMapQueryContext.class, + ci -> new CompiledReplicatedMapQueryContext<>(ci, + ReplicatedChronicleMap.this)); q.initUsed(true); return q; } @@ -489,7 +474,7 @@ public class ReplicatedChronicleMap remoteOpContext() { if (remoteOpContext == null) { - remoteOpContext = q(); + remoteOpContext = (CompiledReplicatedMapQueryContext) q(); } assert !remoteOpContext.usedInit(); remoteOpContext.initUsed(true); @@ -498,7 +483,8 @@ public class ReplicatedChronicleMap remoteItContext() { if (remoteItContext == null) { - remoteItContext = i(); + remoteItContext = + (CompiledReplicatedMapIterationContext) i(); } assert !remoteItContext.usedInit(); remoteItContext.initUsed(true); @@ -518,13 +504,21 @@ public class ReplicatedChronicleMap i() { - //noinspection unchecked - return (CompiledReplicatedMapIterationContext) iterCxt.get(); + private ChainingInterface i() { + ChainingInterface iterContext; + iterContext = cxt.get(); + if (iterContext == null) { + iterContext = new CompiledReplicatedMapIterationContext<>(ReplicatedChronicleMap.this); + cxt.set(iterContext); + } + return iterContext; } public CompiledReplicatedMapIterationContext iterationContext() { - CompiledReplicatedMapIterationContext c = i().getContext(); + CompiledReplicatedMapIterationContext c = + i().getContext(CompiledReplicatedMapIterationContext.class, + ci -> new CompiledReplicatedMapIterationContext<>(ci, + ReplicatedChronicleMap.this)); c.initUsed(true); return c; } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 665391dd7..42fee820f 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.core.io.Closeable; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.VanillaChronicleHash; +import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; @@ -92,9 +93,8 @@ public class VanillaChronicleMap methods; public transient DefaultValueProvider defaultValueProvider; - transient ThreadLocal queryCxt; - transient ThreadLocal iterCxt; - + transient ThreadLocal cxt = new ThreadLocal<>(); + public VanillaChronicleMap(ChronicleMapBuilder builder) throws IOException { super(builder); @@ -162,27 +162,6 @@ public class VanillaChronicleMap valueReader = valueReaderProvider.get(copies, originalValueReader); constantValueProvider.initTransients(valueReader); } - - initQueryContext(); - initIterationContext(); - } - - void initQueryContext() { - queryCxt = new ThreadLocal>() { - @Override - protected CompiledMapQueryContext initialValue() { - return new CompiledMapQueryContext<>(VanillaChronicleMap.this); - } - }; - } - - void initIterationContext() { - iterCxt = new ThreadLocal>() { - @Override - protected CompiledMapIterationContext initialValue() { - return new CompiledMapIterationContext<>(VanillaChronicleMap.this); - } - }; } private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { @@ -314,22 +293,38 @@ public class VanillaChronicleMap q() { - return (CompiledMapQueryContext) queryCxt.get(); + private ChainingInterface q() { + ChainingInterface queryContext; + queryContext = cxt.get(); + if (queryContext == null) { + queryContext = new CompiledMapQueryContext<>(VanillaChronicleMap.this); + cxt.set(queryContext); + } + return queryContext; } public QueryContextInterface mapContext() { - CompiledMapQueryContext q = q().getContext(); + CompiledMapQueryContext q = + q().getContext(CompiledMapQueryContext.class, + ci -> new CompiledMapQueryContext<>(ci, VanillaChronicleMap.this)); q.initUsed(true); return q; } - private CompiledMapIterationContext i() { - return (CompiledMapIterationContext) iterCxt.get(); + private ChainingInterface i() { + ChainingInterface iterContext; + iterContext = cxt.get(); + if (iterContext == null) { + iterContext = new CompiledMapIterationContext<>(VanillaChronicleMap.this); + cxt.set(iterContext); + } + return iterContext; } public IterationContextInterface iterationContext() { - CompiledMapIterationContext c = i().getContext(); + CompiledMapIterationContext c = + i().getContext(CompiledMapIterationContext.class, + ci -> new CompiledMapIterationContext<>(ci, VanillaChronicleMap.this)); c.initUsed(true); return c; } 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 20dfa8e6a..2b231e42c 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData; import net.openhft.chronicle.hash.impl.stage.entry.*; 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.IterationKeyHashCode; @@ -43,6 +44,8 @@ import net.openhft.sg.Staged; ThreadLocalCopiesHolder.class, VanillaChronicleMapHolderImpl.class, + LogHolder.class, + MapSegmentIteration.class, KeyBytesInterop.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 0d7dbde79..8e2e62e99 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java @@ -21,6 +21,7 @@ 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.*; 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.input.HashInputBytes; @@ -44,6 +45,8 @@ import net.openhft.sg.Staged; OwnerThreadHolder.class, ThreadLocalCopiesHolder.class, + LogHolder.class, + VanillaChronicleMapHolderImpl.class, KeyBytesInterop.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 8febc022a..6dbc88cee 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData; import net.openhft.chronicle.hash.impl.stage.entry.*; 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.IterationKeyHashCode; @@ -44,6 +45,8 @@ import net.openhft.sg.Staged; ThreadLocalCopiesHolder.class, ReplicatedChronicleMapHolderImpl.class, + LogHolder.class, + ReplicatedMapSegmentIteration.class, KeyBytesInterop.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java index 903573f49..083772418 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map.impl.stage.map; import net.openhft.chronicle.hash.impl.stage.hash.Chaining; +import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ReplicatedChronicleMap; @@ -38,14 +39,10 @@ public class ReplicatedChronicleMapHolderImpl< this.m = m; } - public ReplicatedChronicleMapHolderImpl(ReplicatedChronicleMapHolderImpl c) { + public ReplicatedChronicleMapHolderImpl( + ChainingInterface c, ReplicatedChronicleMap m) { super(c); - this.m = (ReplicatedChronicleMap) c.m; - } - - @Override - public Chaining createChaining() { - return new ReplicatedChronicleMapHolderImpl(this); + this.m = m; } @Override diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java index 0bdcf1a49..4a637f295 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map.impl.stage.map; import net.openhft.chronicle.hash.impl.stage.hash.Chaining; +import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.VanillaChronicleMap; @@ -38,14 +39,10 @@ public class VanillaChronicleMapHolderImpl< this.m = m; } - public VanillaChronicleMapHolderImpl(VanillaChronicleMapHolderImpl c) { + public VanillaChronicleMapHolderImpl( + ChainingInterface c, VanillaChronicleMap m) { super(c); - this.m = (VanillaChronicleMap) c.m; - } - - @Override - public Chaining createChaining() { - return new VanillaChronicleMapHolderImpl(this); + this.m = m; } @Override 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 index a12eb67ee..26a2b82d6 100644 --- 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 @@ -51,9 +51,10 @@ public abstract class ReplicatedQuerySegmentStages extends QuerySegmentStages { return ret; } else { ReplicatedChronicleMap map = mh.m(); - MapSegmentContext sc = map.segmentContext(segmentIndex); - ((ReplicatedHashSegmentContext) sc) - .forEachSegmentReplicableEntry(cleanupAction); + try (MapSegmentContext sc = map.segmentContext(segmentIndex)) { + ((ReplicatedHashSegmentContext) sc) + .forEachSegmentReplicableEntry(cleanupAction); + } ret = allocReturnCode(chunks); if (ret >= 0) { return ret;