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

Fix several bugs

Browse Source
This commit is contained in:
leventov
2015-09-28 22:13:03 +03:00
parent 440b075f9d
commit c88340f842
18 changed files with 340 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+31 -8
View File
@@ -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<K, KI, MKI extends MetaBytesInterop<K
return actualSegments;
}
/**
* Tier data lock doesn't yet need read-write levels and waits;
* VanillaReadWriteWithWaitsLockingStrategy is just the simplest locking strategy available
*/
static final LockingStrategy tierDataLockingStrategy =
VanillaReadWriteWithWaitsLockingStrategy.instance();
static final TryAcquireOperation<LockingStrategy> tierDataLockTryAcquireOperation =
TryAcquireOperations.lock();
static final
AcquisitionStrategy<LockingStrategy, RuntimeException> 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<K, KI, MKI extends MetaBytesInterop<K
}
private void tierDataLock() {
// TODO this is quick access to locking functionality, eligible only because lock word
// in the segment header has offset = 0
BigSegmentHeader.INSTANCE.writeLock(tierDataAddress() + TIER_DATA_LOCK_OFFSET);
tierDataLockAcquisitionStrategy.acquire(
tierDataLockTryAcquireOperation, tierDataLockingStrategy, Access.nativeAccess(),
null, tierDataAddress() + TIER_DATA_LOCK_OFFSET);
}
private void tierDataUnlock() {
BigSegmentHeader.INSTANCE.writeUnlock(tierDataAddress() + TIER_DATA_LOCK_OFFSET);
tierDataLockingStrategy.unlock(Access.nativeAccess(), null,
tierDataAddress() + TIER_DATA_LOCK_OFFSET);
}
protected long extraAllocatedTierBulkCount() {
@@ -543,9 +559,11 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
}
private void linkFreeTiersList(long alreadyAllocatedBulks) {
long alreadyAllocatedTiers = alreadyAllocatedBulks * numberOfTiersInBulk;
for (long tierIndex = alreadyAllocatedTiers;
tierIndex < alreadyAllocatedTiers + numberOfTiersInBulk - 1; tierIndex++) {
long firstTierIndex = actualSegments + 1 +
alreadyAllocatedBulks * numberOfTiersInBulk;
firstFreeTierIndex(firstTierIndex);
for (long tierIndex = firstTierIndex;
tierIndex < firstTierIndex + numberOfTiersInBulk - 1; tierIndex++) {
long tierBaseAddr = tierIndexToBaseAddr(tierIndex);
long tierDataAddr = tierBaseAddr + segmentHashLookupOuterSize;
TierData.nextTierIndex(tierDataAddr, tierIndex + 1);
@@ -572,7 +590,12 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
tierIndex -= 1;
if (tierIndex < actualSegments)
return msBytesOffset(tierIndex);
return tierBulkData(tierIndex).offset;
long extraTierIndex = tierIndex - actualSegments;
long bulkIndex = extraTierIndex >> log2NumberOfTiersInBulk;
if (bulkIndex >= tierBulkOffsets.size())
mapTiers(bulkIndex);
return tierBulkOffsets.get((int) bulkIndex).offset + tierBulkInnerOffsetToTiers +
(extraTierIndex & (numberOfTiersInBulk - 1)) * segmentSize;
}
private TierBulkData tierBulkData(long tierIndex) {
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/LocksInterface.java
+52
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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();
}
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
+121 -42
View File
@@ -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;
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java
+8 -1
View File
@@ -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);
src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java
+18 -13
View File
@@ -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<Chaining> contextChain;
public final List<ChainingInterface> 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<ChainingInterface> getContextChain() {
return contextChain;
}
public <T> 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> T getContext() {
for (Chaining context : contextChain) {
if (!context.usedInit()) {
@Override
public <T> T getContext(
Class<? extends T> contextClass, Function<ChainingInterface, T> 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);
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java
+30
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.hash.impl.stage.hash;
import java.util.List;
import java.util.function.Function;
public interface ChainingInterface {
List<ChainingInterface> getContextChain();
boolean usedInit();
<T> T getContext(
Class<? extends T> contextClass, Function<ChainingInterface, T> createChaining);
}
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
+2
View File
@@ -54,6 +54,8 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
public long hashLookupEntry = 0;
public abstract boolean hashLookupEntryInit();
public void initHashLookupEntry(long entry) {
hashLookupEntry = entry;
}
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationSegmentStages.java
+2 -1
View File
@@ -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) {
src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java
+1 -1
View File
@@ -50,7 +50,7 @@ public abstract class HashQuery<K> implements HashEntry<K> {
public void dropSearchIfNestedContextsAndPresentHashLookupSlotCheckFailed() {
if (s.locksInit()) {
if (s.concurrentSameThreadContexts &&
s.rootContextOnThisSegment.latestSameThreadSegmentModCount !=
s.rootContextOnThisSegment.latestSameThreadSegmentModCount() !=
s.contextModCount) {
if (ks.keySearchInit()) {
if (ks.searchState == PRESENT) {
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+28 -34
View File
@@ -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<K, KI, MKI extends MetaBytesInterop<K, ? sup
}
}
@Override
void initQueryContext() {
queryCxt = new ThreadLocal<
CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R>>() {
@Override
protected CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R>
initialValue() {
return new CompiledReplicatedMapQueryContext<>(ReplicatedChronicleMap.this);
}
};
}
@Override
void initIterationContext() {
iterCxt = new ThreadLocal<
CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R>>() {
@Override
protected CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R>
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<K, KI, MKI extends MetaBytesInterop<K, ? sup
}
}
private CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R> q() {
//noinspection unchecked
return (CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R>) 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<K, KI, MKI, V, VI, MVI, R> mapContext() {
CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R> q = q().getContext();
CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R> q =
q().getContext(CompiledReplicatedMapQueryContext.class,
ci -> new CompiledReplicatedMapQueryContext<>(ci,
ReplicatedChronicleMap.this));
q.initUsed(true);
return q;
}
@@ -489,7 +474,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
*/
private CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R> remoteOpContext() {
if (remoteOpContext == null) {
remoteOpContext = q();
remoteOpContext = (CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R>) q();
}
assert !remoteOpContext.usedInit();
remoteOpContext.initUsed(true);
@@ -498,7 +483,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
private CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R> remoteItContext() {
if (remoteItContext == null) {
remoteItContext = i();
remoteItContext =
(CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R>) i();
}
assert !remoteItContext.usedInit();
remoteItContext.initUsed(true);
@@ -518,13 +504,21 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
}
}
private CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R> i() {
//noinspection unchecked
return (CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R>) 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<K, KI, MKI, V, VI, MVI, R> iterationContext() {
CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R> c = i().getContext();
CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R> c =
i().getContext(CompiledReplicatedMapIterationContext.class,
ci -> new CompiledReplicatedMapIterationContext<>(ci,
ReplicatedChronicleMap.this));
c.initUsed(true);
return c;
}
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+25 -30
View File
@@ -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<K, KI, MKI extends MetaBytesInterop<K, ? super
public transient MapMethods<K, V, R> methods;
public transient DefaultValueProvider<K, V> defaultValueProvider;
transient ThreadLocal<?> queryCxt;
transient ThreadLocal<?> iterCxt;
transient ThreadLocal<ChainingInterface> cxt = new ThreadLocal<>();
public VanillaChronicleMap(ChronicleMapBuilder<K, V> builder) throws IOException {
super(builder);
@@ -162,27 +162,6 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
BytesReader<V> valueReader = valueReaderProvider.get(copies, originalValueReader);
constantValueProvider.initTransients(valueReader);
}
initQueryContext();
initIterationContext();
}
void initQueryContext() {
queryCxt = new ThreadLocal<CompiledMapQueryContext<K, KI, MKI, V, VI, MVI, R>>() {
@Override
protected CompiledMapQueryContext<K, KI, MKI, V, VI, MVI, R> initialValue() {
return new CompiledMapQueryContext<>(VanillaChronicleMap.this);
}
};
}
void initIterationContext() {
iterCxt = new ThreadLocal<CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R>>() {
@Override
protected CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R> initialValue() {
return new CompiledMapIterationContext<>(VanillaChronicleMap.this);
}
};
}
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
@@ -314,22 +293,38 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
return valueSize;
}
private CompiledMapQueryContext<K, KI, MKI, V, VI, MVI, R> q() {
return (CompiledMapQueryContext<K, KI, MKI, V, VI, MVI, R>) 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<K, V, R> mapContext() {
CompiledMapQueryContext<K, KI, MKI, V, VI, MVI, R> q = q().getContext();
CompiledMapQueryContext<K, KI, MKI, V, VI, MVI, R> q =
q().getContext(CompiledMapQueryContext.class,
ci -> new CompiledMapQueryContext<>(ci, VanillaChronicleMap.this));
q.initUsed(true);
return q;
}
private CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R> i() {
return (CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R>) 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<K, V, ?> iterationContext() {
CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R> c = i().getContext();
CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R> c =
i().getContext(CompiledMapIterationContext.class,
ci -> new CompiledMapIterationContext<>(ci, VanillaChronicleMap.this));
c.initUsed(true);
return c;
}
src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java
+3
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
+3
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java
+3
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java
+4 -7
View File
@@ -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<K, KI, MKI, V, VI, MVI, R> m) {
super(c);
this.m = (ReplicatedChronicleMap<K, KI, MKI, V, VI, MVI, R>) c.m;
}
@Override
public Chaining createChaining() {
return new ReplicatedChronicleMapHolderImpl(this);
this.m = m;
}
@Override
src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java
+4 -7
View File
@@ -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<K, KI, MKI, V, VI, MVI, R> m) {
super(c);
this.m = (VanillaChronicleMap<K, KI, MKI, V, VI, MVI, R>) c.m;
}
@Override
public Chaining createChaining() {
return new VanillaChronicleMapHolderImpl(this);
this.m = m;
}
@Override
src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java
+4 -3
View File
@@ -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;