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

Optimize memory in non-persisted case -- don't zero out all memory eagery; remove global mutable state duplication code in VanillaChronicleHash and ReplicatedChronicleMap; make sleep interruptible in OldEntriesCleanup thread; remove old entries cleanup duplicatation tracking; fix bugs

Browse Source
This commit is contained in:
leventov
2015-10-04 07:57:12 +03:00
parent 01c81ca05b
commit 21d5342048
11 changed files with 257 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java
+38
View File
@@ -0,0 +1,38 @@
/*
* 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;
import net.openhft.lang.model.Byteable;
/**
* <h1>This interface is a private API, don't use in client code</h1>
*
* Excluding global mutable state lock
*/
public interface VanillaGlobalMutableState extends Byteable {
long getAllocatedExtraTierBulks();
void setAllocatedExtraTierBulks(long allocatedExtraTierBulks);
long getFirstFreeTierIndex();
void setFirstFreeTierIndex(long firstFreeTierIndex);
long getExtraTiersInUse();
void setExtraTiersInUse(long extraTiersInUse);
}
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+63 -65
View File
@@ -30,6 +30,7 @@ import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.lang.io.*;
import net.openhft.lang.io.serialization.BytesMarshallableSerializer;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.threadlocal.Provider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,7 +67,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
/////////////////////////////////////////////////
// If the hash was created in the first place, or read from disk
public transient boolean created = false;
public transient boolean createdOrInMemory = false;
/////////////////////////////////////////////////
// Key Data model
@@ -122,7 +123,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
/////////////////////////////////////////////////
// Bytes Store (essentially, the base address) and serialization-dependent offsets
protected transient BytesStore ms;
transient Bytes bytes;
protected transient Bytes bytes;
public static class TierBulkData {
public BytesStore langBytesStore;
@@ -157,13 +158,15 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
protected transient boolean closed = false;
private transient VanillaGlobalMutableState globalMutableState;
public VanillaChronicleHash(ChronicleMapBuilder<K, ?> builder) {
// Version
dataFileVersion = BuildVersion.version();
// Because we are in constructor. If the Hash is not created, deserialization bypasses
// the constructor and created = false
this.created = true;
// Because we are in constructor. If the Hash loaded from persistence file, deserialization
// bypasses the constructor and createdOrInMemory = false
this.createdOrInMemory = true;
@SuppressWarnings("deprecation")
ChronicleHashBuilderPrivateAPI<K> privateAPI = builder.privateAPI();
@@ -221,6 +224,14 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
checksumEntries = privateAPI.checksumEntries();
}
protected VanillaGlobalMutableState createGlobalMutableState() {
return DataValueClasses.newDirectReference(VanillaGlobalMutableState.class);
}
protected VanillaGlobalMutableState globalMutableState() {
return globalMutableState;
}
private long segmentSize() {
long ss = segmentHashLookupOuterSize + TIER_COUNTERS_AREA_SIZE +
segmentFreeListOuterSize + segmentEntrySpaceOuterSize;
@@ -265,6 +276,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
}
private void initOwnTransients() {
globalMutableState = createGlobalMutableState();
keyReaderProvider = Provider.of((Class) originalKeyReader.getClass());
keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass());
if (segmentHashLookupEntrySize == 4) {
@@ -280,6 +292,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
public final void createMappedStoreAndSegments(BytesStore bytesStore) throws IOException {
this.ms = bytesStore;
bytes = ms.bytes();
globalMutableState.bytes(bytes, headerSize + GLOBAL_MUTABLE_STATE_VALUE_OFFSET);
onHeaderCreated();
@@ -287,6 +300,9 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
long segmentHeadersSize = actualSegments * segmentHeaderSize;
segmentsOffset = segmentHeadersOffset + segmentHeadersSize;
if (createdOrInMemory)
zeroOutNewlyMappedChronicleMapBytes();
initTierBulks(bytesStore.size());
}
@@ -297,22 +313,28 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
// file.length() > sizeInBytesWithoutTiers() means there are some tiered segments
Math.max(sizeInBytesWithoutTiers(), file.length()),
BytesMarshallableSerializer.create()));
if (created) {
zeroOutNewlyMappedChronicleMapBytes();
}
}
/**
* newly-extended file contents are not guaranteed to be zero
*/
private void zeroOutNewlyMappedChronicleMapBytes() {
protected void zeroOutNewlyMappedChronicleMapBytes() {
zeroOutGlobalMutableState();
zeroOutSegmentHeaders();
zeroOutFirstSegmentTiers();
}
private void zeroOutGlobalMutableState() {
bytes.zeroOut(headerSize, headerSize + 64, true);
long end = headerSize + globalMutableStateTotalUsedSize();
bytes.zeroOut(headerSize, end, true);
}
protected long globalMutableStateTotalUsedSize() {
return GLOBAL_MUTABLE_STATE_VALUE_OFFSET + globalMutableState().maxSize();
}
private void zeroOutSegmentHeaders() {
bytes.zeroOut(segmentHeadersOffset, segmentsOffset, true);
}
private void zeroOutFirstSegmentTiers() {
@@ -332,6 +354,8 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
tierBulkOffsets = new ArrayList<>();
long tierBulkOffset = roundUpRuntimePageSize(sizeInBytesWithoutTiers());
if (byteStoreSize >= tierBulkOffset + tierBulkSizeInBytes) {
assert !createdOrInMemory : "Chronicle Map is allocated in-memory or created " +
"persisted without extra tier bulks";
TierBulkData firstTierBulkData = new TierBulkData(ms, tierBulkOffset);
tierBulkOffsets.add(firstTierBulkData);
while (true) {
@@ -377,7 +401,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
}
public long mapHeaderInnerSize() {
return headerSize + 64; /* tier data cache line */
return headerSize + globalMutableStateTotalUsedSize();
}
@Override
@@ -391,11 +415,11 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
public final long expectedFileSize() {
long sizeInBytesWithoutTiers = sizeInBytesWithoutTiers();
long extraAllocatedTierBulkCount = extraAllocatedTierBulkCount();
if (extraAllocatedTierBulkCount == 0)
long allocatedExtraTierBulks = globalMutableState.getAllocatedExtraTierBulks();
if (allocatedExtraTierBulks == 0)
return sizeInBytesWithoutTiers;
return roundUpRuntimePageSize(sizeInBytesWithoutTiers) +
extraAllocatedTierBulkCount * tierBulkSizeInBytes;
allocatedExtraTierBulks * tierBulkSizeInBytes;
}
@Override
@@ -495,58 +519,29 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
AcquisitionStrategies.spinLoopOrFail(2, TimeUnit.SECONDS);
private static final long GLOBAL_MUTABLE_STATE_LOCK_OFFSET = 0L;
private static final long EXTRA_ALLOCATED_TIER_BULK_COUNT_OFFSET =
GLOBAL_MUTABLE_STATE_LOCK_OFFSET + 8L;
private static final long FIRST_FREE_TIER_INDEX_OFFSET =
EXTRA_ALLOCATED_TIER_BULK_COUNT_OFFSET + 8L;
private static final long TIERS_IN_USE_OFFSET = FIRST_FREE_TIER_INDEX_OFFSET + 8L;
private static final long GLOBAL_MUTABLE_STATE_VALUE_OFFSET = 8L;
private long globalMutableStateAddress() {
return bytes.address() + headerSize;
}
private void globalMutableStateLock() {
public void globalMutableStateLock() {
globalMutableStateLockAcquisitionStrategy.acquire(
globalMutableStateLockTryAcquireOperation, globalMutableStateLockingStrategy,
Access.nativeAccess(), null,
globalMutableStateAddress() + GLOBAL_MUTABLE_STATE_LOCK_OFFSET);
}
private void globalMutableStateUnlock() {
public void globalMutableStateUnlock() {
globalMutableStateLockingStrategy.unlock(Access.nativeAccess(), null,
globalMutableStateAddress() + GLOBAL_MUTABLE_STATE_LOCK_OFFSET);
}
protected long extraAllocatedTierBulkCount() {
return bytes.readLong(headerSize + EXTRA_ALLOCATED_TIER_BULK_COUNT_OFFSET);
}
private void extraAllocatedTierBulkCount(long extraAllocatedTierBulkCount) {
bytes.writeLong(headerSize + EXTRA_ALLOCATED_TIER_BULK_COUNT_OFFSET,
extraAllocatedTierBulkCount);
}
private long firstFreeTierIndex() {
return bytes.readLong(headerSize + FIRST_FREE_TIER_INDEX_OFFSET);
}
private void firstFreeTierIndex(long firstFreeTierIndex) {
bytes.writeLong(headerSize + FIRST_FREE_TIER_INDEX_OFFSET, firstFreeTierIndex);
}
private long tiersInUse() {
return bytes.readLong(headerSize + TIERS_IN_USE_OFFSET);
}
private void tiersInUse(long tiersInUse) {
bytes.writeLong(headerSize + TIERS_IN_USE_OFFSET, tiersInUse);
}
public long allocateTier(int forSegmentIndex, int tier) {
LOG.debug("Allocate tier for segment # {}, tier {}", forSegmentIndex, tier);
globalMutableStateLock();
try {
long tiersInUse = tiersInUse();
long tiersInUse = globalMutableState.getExtraTiersInUse();
if (tiersInUse == maxExtraTiers) {
throw new IllegalStateException("Attempt to allocate " + (maxExtraTiers + 1) +
"th extra segment tier, " + maxExtraTiers + " is maximum.\n" +
@@ -559,23 +554,20 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
"bad. This might be a DOS attack");
}
while (true) {
long firstFreeTierIndex = firstFreeTierIndex();
long firstFreeTierIndex = globalMutableState.getFirstFreeTierIndex();
if (firstFreeTierIndex > 0) {
tiersInUse(tiersInUse + 1);
globalMutableState.setExtraTiersInUse(tiersInUse + 1);
Bytes allocatedTierBytes = tierBytes(firstFreeTierIndex);
long allocatedTierOffset = tierBytesOffset(firstFreeTierIndex);
long tierBaseAddr = allocatedTierBytes.address() + allocatedTierOffset;
long tierCountersAreaAddr = tierBaseAddr + segmentHashLookupOuterSize;
long nextFreeTierIndex = TierCountersArea.nextTierIndex(tierCountersAreaAddr);
// after reading previously stored nextFreeTierIndex (next linked list entry),
// zero out the whole tier
zeroOutNewlyMappedTier(allocatedTierBytes, allocatedTierOffset);
// now, when this tier will be a part of the map, the next tier designates
// the next tier in the data structure, should be 0
TierCountersArea.nextTierIndex(tierCountersAreaAddr, 0);
TierCountersArea.segmentIndex(tierCountersAreaAddr, forSegmentIndex);
TierCountersArea.tier(tierCountersAreaAddr, tier);
firstFreeTierIndex(nextFreeTierIndex);
globalMutableState.setFirstFreeTierIndex(nextFreeTierIndex);
return firstFreeTierIndex;
} else {
allocateTierBulk();
@@ -587,25 +579,31 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
}
private void allocateTierBulk() {
long alreadyAllocatedBulks = extraAllocatedTierBulkCount();
long alreadyAllocatedBulks = globalMutableState.getAllocatedExtraTierBulks();
mapTiers(alreadyAllocatedBulks);
extraAllocatedTierBulkCount(alreadyAllocatedBulks + 1);
linkNewlyAllocatedTiersIntoFreeTiersList(alreadyAllocatedBulks);
}
private void linkNewlyAllocatedTiersIntoFreeTiersList(long alreadyAllocatedBulks) {
long firstTierIndex = actualSegments + 1 +
alreadyAllocatedBulks * numberOfTiersInBulk;
firstFreeTierIndex(firstTierIndex);
long firstTierIndex = actualSegments + 1 + alreadyAllocatedBulks * numberOfTiersInBulk;
Bytes bytes = tierBytes(firstTierIndex);
long firstTierOffset = tierBytesOffset(firstTierIndex);
if (tierBulkInnerOffsetToTiers > 0) {
// These bytes are bit sets in Replicated version
bytes.zeroOut(firstTierOffset - tierBulkInnerOffsetToTiers, firstTierOffset, true);
}
// Link newly allocated tiers into free tiers list
for (long tierIndex = firstTierIndex;
tierIndex < firstTierIndex + numberOfTiersInBulk - 1; tierIndex++) {
long tierOffset = tierBytesOffset(tierIndex);
zeroOutNewlyMappedTier(bytes, tierOffset);
long tierCountersAreaOffset = tierOffset + segmentHashLookupOuterSize;
// zero out tier counters area, because newly mapped memory could be dirty
bytes.zeroOut(tierCountersAreaOffset, tierCountersAreaOffset + TIER_COUNTERS_AREA_SIZE);
TierCountersArea.nextTierIndex(bytes.address() + tierCountersAreaOffset, tierIndex + 1);
}
// TODO HCOLL-397 insert msync here!
// after we are sure the new bulk is initialized, update the global mutable state
globalMutableState.setAllocatedExtraTierBulks(alreadyAllocatedBulks + 1);
globalMutableState.setFirstFreeTierIndex(firstTierIndex);
}
public long tierIndexToBaseAddr(long tierIndex) {
@@ -687,7 +685,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
int firstBulkToMap = tierBulkOffsets.size();
long bulksToMap = upToBulkIndex + 1 - firstBulkToMap;
long mapSize = bulksToMap * tierBulkSizeInBytes;
DirectStore extraStore = new DirectStore(ms.objectSerializer(), mapSize, true);
DirectStore extraStore = new DirectStore(ms.objectSerializer(), mapSize, false);
appendTierBulkData(upToBulkIndex, firstBulkToMap, extraStore);
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+17 -39
View File
@@ -1626,7 +1626,7 @@ public final class ChronicleMapBuilder<K, V> implements
// throw new IllegalStateException("Windows cannot support this configuration");
// }
BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE,
map.sizeInBytesWithoutTiers(), true);
map.sizeInBytesWithoutTiers(), false);
map.createMappedStoreAndSegments(bytesStore);
return establishReplication(map, singleHashReplication, channel);
} catch (IOException e) {
@@ -1727,46 +1727,24 @@ public final class ChronicleMapBuilder<K, V> implements
}
private void establishCleanupThread(ReplicatedChronicleMap map) {
map.globalStateLock();
try {
if (!map.globalMutableState.isCurrentlyCleanupIterated()) {
OldDeletedEntriesCleanup cleanup = new OldDeletedEntriesCleanup(map);
NamedThreadFactory threadFactory =
new NamedThreadFactory("cleanup thread for map persisted at " + map.file());
ExecutorService executor = Executors.newSingleThreadExecutor(threadFactory);
executor.submit(cleanup);
OldDeletedEntriesCleanup cleanup = new OldDeletedEntriesCleanup(map);
NamedThreadFactory threadFactory =
new NamedThreadFactory("cleanup thread for map persisted at " + map.file());
ExecutorService executor = Executors.newSingleThreadExecutor(threadFactory);
executor.submit(cleanup);
map.addCloseable(cleanup);
// WARNING this relies on the fact that ReplicatedChronicleMap closes in the same
// order, i. e. OldDeletedEntriesCleanup instance first
map.addCloseable(() -> {
executor.shutdown();
try {
executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOG.error("", e);
}
});
map.addCloseable(() -> {
map.globalStateLock();
try {
map.globalMutableState.setCurrentlyCleanupIterated(false);
} finally {
map.globalStateUnlock();
}
});
map.globalStateLock();
try {
if (!map.globalMutableState.isCurrentlyCleanupIterated())
map.globalMutableState.setCurrentlyCleanupIterated(true);
} finally {
map.globalStateUnlock();
}
map.addCloseable(cleanup);
// WARNING this relies on the fact that ReplicatedChronicleMap closes closeables in the same
// order as they are added, i. e. OldDeletedEntriesCleanup instance close()d before the
// following closeable
map.addCloseable(() -> {
executor.shutdown();
try {
executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOG.error("", e);
}
} finally {
map.globalStateUnlock();
}
});
}
private long bufferSize(SerializationBuilder builder, double averageSize) {
src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java
+41 -26
View File
@@ -26,6 +26,7 @@ import org.slf4j.LoggerFactory;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Predicate;
public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<ReplicableEntry> {
@@ -35,6 +36,7 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<
private final int[] segmentsPermutation;
private final int[] inverseSegmentsPermutation;
private volatile boolean shutdown;
private volatile Thread runnerThread;
private long prevSegment0ScanStart = -1;
private long removedCompletely;
@@ -46,10 +48,28 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<
@Override
public void run() {
runnerThread = Thread.currentThread();
while (!shutdown) {
int segmentIndex = map.globalMutableState.getCurrentCleanupSegmentIndex();
int segmentIndex = map.globalMutableState().getCurrentCleanupSegmentIndex();
int nextSegmentIndex;
TimeProvider timeProvider = map.timeProvider;
if (segmentIndex == 0 && prevSegment0ScanStart >= 0) {
try (MapSegmentContext<?, ?, ?> context = map.segmentContext(segmentIndex)) {
if (segmentIndex == 0)
prevSegment0ScanStart = timeProvider.currentTime();
removedCompletely = 0;
if (((ReplicatedHashSegmentContext<?, ?>) context)
.forEachSegmentReplicableEntryWhile(this)) {
LOG.debug("Removed {} old deleted entries in the segment {}",
removedCompletely, segmentIndex);
nextSegmentIndex = nextSegmentIndex(segmentIndex);
map.globalMutableState().setCurrentCleanupSegmentIndex(nextSegmentIndex);
} else {
// forEachWhile returned false => interrupted => shutdown = true
assert shutdown;
return;
}
}
if (nextSegmentIndex == 0) {
long currentTime = timeProvider.currentTime();
TimeUnit cleanupTimeoutUnit = map.cleanupTimeoutUnit;
long mapScanTime = timeProvider.systemTimeIntervalBetween(
@@ -59,34 +79,12 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<
if (mapScanTime < cleanupTimeout) {
long timeToSleep = cleanupTimeoutUnit.toMillis(cleanupTimeout - mapScanTime);
if (timeToSleep > 0) {
try {
Thread.sleep(timeToSleep);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
sleepMillis(timeToSleep);
} else {
timeToSleep = cleanupTimeoutUnit.toNanos(cleanupTimeout - mapScanTime);
try {
Thread.sleep(0, (int) timeToSleep);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
sleepNanos(cleanupTimeoutUnit.toNanos(cleanupTimeout - mapScanTime));
}
}
}
try (MapSegmentContext<?, ?, ?> context = map.segmentContext(segmentIndex)) {
if (segmentIndex == 0)
prevSegment0ScanStart = timeProvider.currentTime();
removedCompletely = 0;
if (((ReplicatedHashSegmentContext<?, ?>) context)
.forEachSegmentReplicableEntryWhile(this)) {
segmentIndex = nextSegmentIndex(segmentIndex);
map.globalMutableState.setCurrentCleanupSegmentIndex(segmentIndex);
}
}
LOG.debug("Removed {} old deleted entries in the segment {}",
removedCompletely, segmentIndex);
}
}
@@ -105,9 +103,26 @@ public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<
return true;
}
private void sleepMillis(long millis) {
long deadline = System.currentTimeMillis() + millis;
while (System.currentTimeMillis() < deadline && !shutdown)
LockSupport.parkUntil(this, deadline);
}
private void sleepNanos(long nanos) {
long deadline = System.nanoTime() + nanos;
while (System.nanoTime() < deadline && !shutdown)
LockSupport.parkNanos(this, deadline);
}
@Override
public void close() {
shutdown = true;
if (runnerThread != null &&
// this means blocked in sleepMillis() or sleepNanos()
LockSupport.getBlocker(runnerThread) == this) {
runnerThread.interrupt(); // unblock
}
}
private int nextSegmentIndex(int segmentIndex) {
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+20 -23
View File
@@ -19,6 +19,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.algo.bitset.BitSetFrame;
import net.openhft.chronicle.algo.bitset.ConcurrentFlatBitSetFrame;
import net.openhft.chronicle.algo.bitset.SingleThreadedFlatBitSetFrame;
import net.openhft.chronicle.hash.VanillaGlobalMutableState;
import net.openhft.chronicle.hash.impl.TierCountersArea;
import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface;
@@ -99,14 +100,11 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
private static final Logger LOG = LoggerFactory.getLogger(ReplicatedChronicleMap.class);
private static final long LAST_UPDATED_HEADER_SIZE = 128L * 8L;
private static final int SIZE_OF_BOOTSTRAP_TIME_STAMP = 8;
private static final long GLOBAL_MUTABLE_STATE_SIZE = 64; // a cache line
public final TimeProvider timeProvider;
private final byte localIdentifier;
transient Set<Closeable> closeables;
private transient Bytes identifierUpdatedBytes;
transient Bytes globalMutableStateBytes;
transient ReplicatedChronicleMapGlobalMutableState globalMutableState;
private transient ATSDirectBitSet modIterSet;
private transient AtomicReferenceArray<ModificationIterator> modificationIterators;
@@ -143,6 +141,16 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
}
}
@Override
protected VanillaGlobalMutableState createGlobalMutableState() {
return DataValueClasses.newDirectReference(ReplicatedGlobalMutableState.class);
}
@Override
protected ReplicatedGlobalMutableState globalMutableState() {
return (ReplicatedGlobalMutableState) super.globalMutableState();
}
private int assignedModIterBitSetSizeInBytes() {
return (int) CACHE_LINES.align(BYTES.alignAndConvert(127 + RESERVED_MOD_ITER, BITS), BYTES);
}
@@ -197,9 +205,9 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
@Override
public long mapHeaderInnerSize() {
return super.mapHeaderInnerSize() + GLOBAL_MUTABLE_STATE_SIZE + LAST_UPDATED_HEADER_SIZE +
(modIterBitSetSizeInBytes() * (128 + RESERVED_MOD_ITER)) +
assignedModIterBitSetSizeInBytes();
return super.mapHeaderInnerSize() + LAST_UPDATED_HEADER_SIZE +
assignedModIterBitSetSizeInBytes() +
(modIterBitSetSizeInBytes() * (128 + RESERVED_MOD_ITER));
}
public void setLastModificationTime(byte identifier, long timestamp) {
@@ -224,12 +232,6 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
public void onHeaderCreated() {
long offset = super.mapHeaderInnerSize();
globalMutableStateBytes = ms.bytes(offset, GLOBAL_MUTABLE_STATE_SIZE);
globalMutableState =
DataValueClasses.newDirectReference(ReplicatedChronicleMapGlobalMutableState.class);
globalMutableState.bytes(globalMutableStateBytes, 8);
offset += GLOBAL_MUTABLE_STATE_SIZE;
identifierUpdatedBytes = ms.bytes(offset, LAST_UPDATED_HEADER_SIZE);
offset += LAST_UPDATED_HEADER_SIZE;
@@ -239,16 +241,10 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
modIterSet = new ATSDirectBitSet(modDelBytes);
}
void globalStateLock() {
try {
globalMutableStateBytes.busyLockLong(0);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
void globalStateUnlock() {
globalMutableStateBytes.unlockLong(0);
@Override
protected void zeroOutNewlyMappedChronicleMapBytes() {
super.zeroOutNewlyMappedChronicleMapBytes();
bytes.zeroOut(super.mapHeaderInnerSize(), this.mapHeaderInnerSize(), true);
}
void addCloseable(Closeable closeable) {
@@ -701,7 +697,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
}
}
// for each allocated tier bulk
while (iterationMainSegmentsAreaOrTierBulk < extraAllocatedTierBulkCount()) {
while (iterationMainSegmentsAreaOrTierBulk <
globalMutableState().getAllocatedExtraTierBulks()) {
VanillaChronicleHash.TierBulkData tierBulkData =
tierBulkOffsets.get(iterationMainSegmentsAreaOrTierBulk);
tierBulkBitSetAddr = tierBulkData.langBytes.address() + tierBulkData.offset +
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMapGlobalMutableState.java → src/main/java/net/openhft/chronicle/map/ReplicatedGlobalMutableState.java
+2 -6
View File
@@ -16,15 +16,11 @@
package net.openhft.chronicle.map;
import net.openhft.lang.model.Byteable;
import net.openhft.chronicle.hash.VanillaGlobalMutableState;
interface ReplicatedChronicleMapGlobalMutableState extends Byteable {
interface ReplicatedGlobalMutableState extends VanillaGlobalMutableState {
int getCurrentCleanupSegmentIndex();
void setCurrentCleanupSegmentIndex(int currentCleanupSegmentIndex);
boolean isCurrentlyCleanupIterated();
void setCurrentlyCleanupIterated(boolean isCurrentlyCleanupIterated);
}