From 21d5342048d40adb06cc0f9deea50c6c5975837b Mon Sep 17 00:00:00 2001 From: leventov Date: Sun, 4 Oct 2015 07:56:57 +0300 Subject: [PATCH] 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 --- pom.xml | 1 + .../hash/VanillaGlobalMutableState.java | 38 ++++++ .../hash/impl/VanillaChronicleHash.java | 128 +++++++++--------- .../chronicle/map/ChronicleMapBuilder.java | 56 +++----- .../map/OldDeletedEntriesCleanup.java | 67 +++++---- .../chronicle/map/ReplicatedChronicleMap.java | 43 +++--- ...java => ReplicatedGlobalMutableState.java} | 8 +- .../java/eg/TestNesterInterfaceMarshal.java | 9 +- .../net/openhft/chronicle/map/Builder.java | 25 +++- .../chronicle/map/GlobalMutableStateTest.java | 48 +++++++ .../map/TimeBasedReplicationTest.java | 14 +- 11 files changed, 257 insertions(+), 180 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java rename src/main/java/net/openhft/chronicle/map/{ReplicatedChronicleMapGlobalMutableState.java => ReplicatedGlobalMutableState.java} (79%) create mode 100644 src/test/java/net/openhft/chronicle/map/GlobalMutableStateTest.java diff --git a/pom.xml b/pom.xml index 1780500b4..05265a34d 100755 --- a/pom.xml +++ b/pom.xml @@ -230,6 +230,7 @@ net/openhft/chronicle/hash/ChronicleHashBuilderPrivateAPI.java + net/openhft/chronicle/hash/VanillaGlobalMutableState.java net/openhft/chronicle/hash/Segment.java net/openhft/chronicle/map/ConstantValueProvider.java net/openhft/chronicle/map/VanillaChronicleMap.java diff --git a/src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java b/src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java new file mode 100644 index 000000000..e658efee3 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java @@ -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 . + */ + +package net.openhft.chronicle.hash; + +import net.openhft.lang.model.Byteable; + +/** + *

This interface is a private API, don't use in client code

+ * + * 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); +} 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 69c0bd221..104842592 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -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 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 privateAPI = builder.privateAPI(); @@ -221,6 +224,14 @@ public abstract class VanillaChronicleHash 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(); 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 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 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 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 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) { diff --git a/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java index c363917fd..944c2a524 100644 --- a/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java +++ b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java @@ -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 { @@ -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) { diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index f1ad42757..b750ccf8e 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -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 closeables; private transient Bytes identifierUpdatedBytes; - transient Bytes globalMutableStateBytes; - transient ReplicatedChronicleMapGlobalMutableState globalMutableState; private transient ATSDirectBitSet modIterSet; private transient AtomicReferenceArray modificationIterators; @@ -143,6 +141,16 @@ public class ReplicatedChronicleMap timeOutMs) + break; } } diff --git a/src/test/java/net/openhft/chronicle/map/GlobalMutableStateTest.java b/src/test/java/net/openhft/chronicle/map/GlobalMutableStateTest.java new file mode 100644 index 000000000..fd05cbd11 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/GlobalMutableStateTest.java @@ -0,0 +1,48 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.VanillaGlobalMutableState; +import net.openhft.lang.model.DataValueClasses; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class GlobalMutableStateTest { + + private String dumpCode; + + @Before + public void setDumpCode() { + dumpCode = System.getProperty("dvg.dumpCode"); + System.setProperty("dvg.dumpCode", "true"); + } + + @After + public void unsetDumpCode() { + if (dumpCode != null) + System.setProperty("dvg.dumpCode", dumpCode); + else + System.getProperties().remove("dvg.dumpCode"); + } + + @Test + public void globalMutableStateTest() { + DataValueClasses.newDirectReference(ReplicatedGlobalMutableState.class); + DataValueClasses.newDirectReference(VanillaGlobalMutableState.class); + } +} diff --git a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java index d79faf284..642ea31b7 100644 --- a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java @@ -55,7 +55,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase { public void testIgnoreLateAction(BiConsumer, ChronicleMap> action) - throws IOException { + throws IOException, InterruptedException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); try (ChronicleMap map = @@ -97,7 +97,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase { } @Test - public void testIgnoreALatePut() throws IOException { + public void testIgnoreALatePut() throws IOException, InterruptedException { testIgnoreLateAction((m1, m2) -> { m2.put("key-1", "value-2"); assertEquals("value-2", m2.get("key-1")); @@ -137,7 +137,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase { } @Test - public void testIgnoreALateReplace() throws IOException { + public void testIgnoreALateReplace() throws IOException, InterruptedException { testIgnoreLateAction((m1, m2) -> { // in order to be able to call replace() "locally" m2.put("key-1", "dummy"); @@ -147,7 +147,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase { } @Test - public void testIgnoreALateReplaceWithValue() throws IOException { + public void testIgnoreALateReplaceWithValue() throws IOException, InterruptedException { testIgnoreLateAction((m1, m2) -> { // in order to be able to call replace() "locally" m2.put("key-1", "dummy"); @@ -157,7 +157,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase { } @Test - public void testIgnoreALateRemoveWithValue() throws IOException { + public void testIgnoreALateRemoveWithValue() throws IOException, InterruptedException { testIgnoreLateAction((m1, m2) -> { // in order to be able to call remove() "locally" m2.put("key-1", "dummy"); @@ -167,7 +167,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase { } @Test - public void testIgnoreALateRemove() throws IOException { + public void testIgnoreALateRemove() throws IOException, InterruptedException { testIgnoreLateAction((m1, m2) -> { // in order to be able to call remove() "locally" m2.put("key-1", "dummy"); @@ -177,7 +177,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase { } @Test - public void testRemoveFollowedByLatePut() throws IOException { + public void testRemoveFollowedByLatePut() throws IOException, InterruptedException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); try (ChronicleMap map =