From 9a2bb31b6620b8ce0b43e1d315716d801a0d9ba5 Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 5 Nov 2016 15:08:45 -0600 Subject: [PATCH] Add ChronicleHash.offHeapMemoryUsed() method; Centralized management of all direct memory and file resources of ChronicleMap in one Cleaner --- .../openhft/chronicle/hash/ChronicleHash.java | 10 ++ .../impl/ChronicleHashResourceReleaser.java | 59 +++++++++ ...InMemoryChronicleHashResourceReleaser.java | 71 ++++++++++ ...erenceCounted.java => MemoryResource.java} | 21 +-- ...ersistedChronicleHashResourceReleaser.java | 89 +++++++++++++ .../hash/impl/VanillaChronicleHash.java | 91 ++++++------- .../chronicle/map/ChronicleMapBuilder.java | 109 +++++++++------- .../chronicle/map/VanillaChronicleMap.java | 4 +- .../net/openhft/chronicle/set/SetFromMap.java | 5 + .../openhft/chronicle/map/MemoryLeakTest.java | 50 ------- .../chronicle/map/MemoryResourcesTest.java | 123 ++++++++++++++++++ .../chronicle/map/ReplicationCheckingMap.java | 5 + 12 files changed, 478 insertions(+), 159 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResourceReleaser.java create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResourceReleaser.java rename src/main/java/net/openhft/chronicle/hash/impl/{DummyReferenceCounted.java => MemoryResource.java} (69%) create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResourceReleaser.java delete mode 100644 src/test/java/net/openhft/chronicle/map/MemoryLeakTest.java create mode 100644 src/test/java/net/openhft/chronicle/map/MemoryResourcesTest.java diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java index a985866b7..23e82cd0f 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java @@ -52,6 +52,16 @@ public interface ChronicleHash, SC extends HashSegment */ long longSize(); + /** + * Returns the amount of off-heap memory (in bytes), allocated by this {@code ChronicleHash} or + * shared with with other ChronicleHashes, persisting to the same {@link #file()}. + * + *

After {@link #close()} this method returns 0. + * + * @return the amount of off-heap memory, used by this {@code ChronicleHash} (in bytes) + */ + long offHeapMemoryUsed(); + /** * @return the class of {@code } */ diff --git a/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResourceReleaser.java b/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResourceReleaser.java new file mode 100644 index 000000000..8519fc29c --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResourceReleaser.java @@ -0,0 +1,59 @@ +/* + * Copyright (C) 2012, 2016 higherfrequencytrading.com + * Copyright (C) 2016 Roman Leventov + * + * 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; + +import net.openhft.chronicle.hash.impl.util.Throwables; + +import java.util.ArrayList; +import java.util.List; + +public abstract class ChronicleHashResourceReleaser implements Runnable { + + List memoryResources = new ArrayList<>(1); + + public final synchronized void addMemoryResource(long address, long size) { + if (memoryResources == null) + throw new IllegalStateException("Already released"); + memoryResources.add(new MemoryResource(address, size)); + } + + public final synchronized long totalMemory() { + if (memoryResources == null) + return 0L; + long totalMemory = 0L; + for (MemoryResource memoryResource : memoryResources) { + totalMemory += memoryResource.size; + } + return totalMemory; + } + + abstract Throwable doRelease(); + + public final synchronized void releaseManually() { + if (memoryResources == null) + return; // Already released + Throwable thrown = null; + try { + thrown = doRelease(); + } finally { + memoryResources = null; + } + if (thrown != null) + throw Throwables.propagate(thrown); + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResourceReleaser.java b/src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResourceReleaser.java new file mode 100644 index 000000000..ae2a51de4 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResourceReleaser.java @@ -0,0 +1,71 @@ +/* + * Copyright (C) 2012, 2016 higherfrequencytrading.com + * Copyright (C) 2016 Roman Leventov + * + * 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; + +import net.openhft.chronicle.core.OS; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public final class InMemoryChronicleHashResourceReleaser extends ChronicleHashResourceReleaser { + + private static final Logger LOG = + LoggerFactory.getLogger(PersistedChronicleHashResourceReleaser.class); + + public InMemoryChronicleHashResourceReleaser() {} + + @Override + public synchronized void run() { + if (memoryResources == null) + return; // Already released + // All Throwables are catched because this class is used as Runnable for sun.misc.Cleaner, + // hence must not fail + Throwable thrown = null; + try { + thrown = doRelease(); + } finally { + memoryResources = null; + } + if (thrown != null) { + try { + LOG.error("Error on releasing memory of a Chronicle Map:", thrown); + } catch (Throwable t) { + // This may occur if Releaser is run in a shutdown hook, and the log service has + // already been shut down. Try to fall back to printStackTrace(). + thrown.addSuppressed(t); + thrown.printStackTrace(); + } + } + } + + @Override + Throwable doRelease() { + Throwable thrown = null; + for (MemoryResource allocation : memoryResources) { + try { + OS.memory().freeMemory(allocation.address, allocation.size); + } catch (Throwable t) { + if (thrown == null) { + thrown = t; + } else { + thrown.addSuppressed(t); + } + } + } + return thrown; + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/impl/DummyReferenceCounted.java b/src/main/java/net/openhft/chronicle/hash/impl/MemoryResource.java similarity index 69% rename from src/main/java/net/openhft/chronicle/hash/impl/DummyReferenceCounted.java rename to src/main/java/net/openhft/chronicle/hash/impl/MemoryResource.java index de831f950..18993fa1e 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/DummyReferenceCounted.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/MemoryResource.java @@ -17,21 +17,12 @@ package net.openhft.chronicle.hash.impl; -import net.openhft.chronicle.core.ReferenceCounted; +final class MemoryResource { + final long address; + final long size; -public enum DummyReferenceCounted implements ReferenceCounted { - DUMMY_REFERENCE_COUNTED; - - @Override - public void reserve() throws IllegalStateException { - } - - @Override - public void release() throws IllegalStateException { - } - - @Override - public long refCount() { - return 0; + MemoryResource(long address, long size) { + this.address = address; + this.size = size; } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResourceReleaser.java b/src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResourceReleaser.java new file mode 100644 index 000000000..de04c6cdd --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResourceReleaser.java @@ -0,0 +1,89 @@ +/* + * Copyright (C) 2012, 2016 higherfrequencytrading.com + * Copyright (C) 2016 Roman Leventov + * + * 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; + +import net.openhft.chronicle.core.OS; +import net.openhft.chronicle.hash.impl.util.CanonicalRandomAccessFiles; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; + + +public final class PersistedChronicleHashResourceReleaser + extends ChronicleHashResourceReleaser { + private static final Logger LOG = + LoggerFactory.getLogger(PersistedChronicleHashResourceReleaser.class); + + private final File file; + + public PersistedChronicleHashResourceReleaser(File file) { + this.file = file; + } + + @Override + public synchronized void run() { + if (memoryResources == null) + return; // Already released + // All Throwables are catched because this class is used as Runnable for sun.misc.Cleaner, + // hence must not fail + Throwable thrown = null; + try { + thrown = doRelease(); + } finally { + memoryResources = null; + } + if (thrown != null) { + try { + LOG.error("Error on releasing memory and RAF of a Chronicle Map at file=" + file, + thrown); + } catch (Throwable t) { + // This may occur if Releaser is run in a shutdown hook, and the log service has + // already been shut down. Try to fall back to printStackTrace(). + thrown.addSuppressed(t); + thrown.printStackTrace(); + } + } + } + + @Override + Throwable doRelease() { + Throwable thrown = null; + for (MemoryResource mapping : memoryResources) { + try { + OS.unmap(mapping.address, mapping.size); + } catch (Throwable t) { + if (thrown == null) { + thrown = t; + } else { + thrown.addSuppressed(t); + } + } + } + try { + CanonicalRandomAccessFiles.release(file); + } catch (Throwable t) { + if (thrown == null) { + thrown = t; + } else { + thrown.addSuppressed(t); + } + } + return thrown; + } +} 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 4582481d2..bb80b528b 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -22,11 +22,11 @@ import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.bytes.MappedBytesStoreFactory; import net.openhft.chronicle.bytes.NativeBytesStore; import net.openhft.chronicle.core.Maths; +import net.openhft.chronicle.core.Memory; import net.openhft.chronicle.core.OS; import net.openhft.chronicle.hash.*; import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.hash.impl.util.BuildVersion; -import net.openhft.chronicle.hash.impl.util.CanonicalRandomAccessFiles; import net.openhft.chronicle.hash.impl.util.jna.PosixMsync; import net.openhft.chronicle.hash.impl.util.jna.WindowsMsync; import net.openhft.chronicle.hash.serialization.DataAccess; @@ -57,10 +57,8 @@ import static java.lang.Math.max; import static java.nio.channels.FileChannel.MapMode.READ_WRITE; import static net.openhft.chronicle.algo.MemoryUnit.*; import static net.openhft.chronicle.algo.bytes.Access.nativeAccess; -import static net.openhft.chronicle.bytes.NativeBytesStore.lazyNativeBytesStoreWithFixedCapacity; import static net.openhft.chronicle.core.OS.pageAlign; import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.*; -import static net.openhft.chronicle.hash.impl.DummyReferenceCounted.DUMMY_REFERENCE_COUNTED; public abstract class VanillaChronicleHash, SC extends HashSegmentContext, @@ -131,7 +129,8 @@ public abstract class VanillaChronicleHash GIGABYTES.toBytes(1)) { @@ -681,16 +674,10 @@ public abstract class VanillaChronicleHash 0; + } + + @Override + public long offHeapMemoryUsed() { + return resourceReleaser.totalMemory(); + } + public long allocateTier() { globalMutableStateLock(); try { @@ -992,8 +989,8 @@ public abstract class VanillaChronicleHash(address, capacity, null, false); } private void appendBulkData(int firstBulkToMapIndex, int upToBulkIndex, BytesStore extraStore, diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 223f023a8..a861e35f2 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -22,15 +22,15 @@ import net.openhft.chronicle.algo.hashing.LongHashFunction; import net.openhft.chronicle.bytes.Byteable; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.bytes.BytesStore; +import net.openhft.chronicle.bytes.NativeBytesStore; import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.core.OS; import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashRecoveryFailedException; -import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; -import net.openhft.chronicle.hash.impl.SizePrefixedBlob; -import net.openhft.chronicle.hash.impl.VanillaChronicleHash; +import net.openhft.chronicle.hash.impl.*; import net.openhft.chronicle.hash.impl.stage.entry.ChecksumStrategy; import net.openhft.chronicle.hash.impl.util.CanonicalRandomAccessFiles; +import net.openhft.chronicle.hash.impl.util.Throwables; import net.openhft.chronicle.hash.impl.util.math.PoissonDistribution; import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.hash.serialization.impl.SerializationBuilder; @@ -60,7 +60,6 @@ import java.util.concurrent.TimeUnit; import static java.lang.Double.isNaN; import static java.lang.Math.round; import static java.nio.ByteOrder.LITTLE_ENDIAN; -import static net.openhft.chronicle.bytes.NativeBytesStore.lazyNativeBytesStoreWithFixedCapacity; import static net.openhft.chronicle.core.Maths.*; import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.*; import static net.openhft.chronicle.hash.impl.SizePrefixedBlob.*; @@ -215,7 +214,7 @@ public final class ChronicleMapBuilder implements /** * When Chronicle Maps are created using {@link #createPersistedTo(File)} or * {@link #recoverPersistedTo(File, boolean)} or {@link #createOrRecoverPersistedTo(File)} - * methods, file lock on the Chronicle Map's lock is acquired, that shouldn't be done from + * methods, file lock on the Chronicle Map's file is acquired, that shouldn't be done from * concurrent threads within the same JVM process. So creation of Chronicle Maps * persisted to the same File should be synchronized across JVM's threads. Simple way would be * to synchronize on some static (lock) object, but would serialize all Chronicle Maps creations @@ -1472,46 +1471,57 @@ public final class ChronicleMapBuilder implements //noinspection ResultOfMethodCallIgnored file.createNewFile(); } - VanillaChronicleMap result = null; RandomAccessFile raf = CanonicalRandomAccessFiles.acquire(file); + ChronicleHashResourceReleaser resourceReleaser = + new PersistedChronicleHashResourceReleaser(file); try { - if (raf.length() > 0) - return result = openWithExistingFile(file, raf, recover, overrideBuilderConfig); + VanillaChronicleMap result; + if (raf.length() > 0) { + result = openWithExistingFile(file, raf, resourceReleaser, recover, + overrideBuilderConfig); + } else { - // Single-element arrays allow to modify variables within lambda - @SuppressWarnings("unchecked") - VanillaChronicleMap[] map = new VanillaChronicleMap[1]; - ByteBuffer[] headerBuffer = new ByteBuffer[1]; - boolean[] newFile = new boolean[1]; - FileChannel fileChannel = raf.getChannel(); + // Single-element arrays allow to modify variables within lambda + @SuppressWarnings("unchecked") + VanillaChronicleMap[] map = new VanillaChronicleMap[1]; + ByteBuffer[] headerBuffer = new ByteBuffer[1]; + boolean[] newFile = new boolean[1]; + FileChannel fileChannel = raf.getChannel(); - fileLockedIO(file, fileChannel, () -> { - if (raf.length() == 0) { - map[0] = newMap(); - headerBuffer[0] = writeHeader(fileChannel, map[0]); - newFile[0] = true; + fileLockedIO(file, fileChannel, () -> { + if (raf.length() == 0) { + map[0] = newMap(); + headerBuffer[0] = writeHeader(fileChannel, map[0]); + newFile[0] = true; + } else { + newFile[0] = false; + } + }); + + if (newFile[0]) { + int headerSize = headerBuffer[0].remaining(); + result = createWithNewFile(map[0], file, raf, resourceReleaser, headerBuffer[0], + headerSize); } else { - newFile[0] = false; + result = openWithExistingFile(file, raf, resourceReleaser, recover, + overrideBuilderConfig); } - }); - - if (newFile[0]) { - int headerSize = headerBuffer[0].remaining(); - return result = createWithNewFile(map[0], file, raf, headerBuffer[0], headerSize); - } else { - return result = openWithExistingFile(file, raf, recover, overrideBuilderConfig); } - } finally { - if (result != null) { - result.registerRafReleaser(); - prepareMapPublication(result); - } else { - CanonicalRandomAccessFiles.release(file); + prepareMapPublication(result); + return result; + } catch (Throwable t) { + try { + resourceReleaser.releaseManually(); + } catch (Exception e) { + t.addSuppressed(e); } + throw Throwables.propagateNotWrapping(t, IOException.class); } } - private static void prepareMapPublication(VanillaChronicleMap map) { + private void prepareMapPublication(VanillaChronicleMap map) throws IOException { + establishReplication(map); + map.registerCleaner(); // Ensure safe publication of a ChronicleMap OS.memory().storeFence(); ChronicleMapCloseOnExitHook.add(map); @@ -1607,16 +1617,17 @@ public final class ChronicleMapBuilder implements private VanillaChronicleMap createWithNewFile( VanillaChronicleMap map, File file, RandomAccessFile raf, - ByteBuffer headerBuffer, int headerSize) throws IOException { + ChronicleHashResourceReleaser resourceReleaser, ByteBuffer headerBuffer, int headerSize) + throws IOException { map.initBeforeMapping(file, raf, headerBuffer.limit(), false); - map.createMappedStoreAndSegments(); + map.createMappedStoreAndSegments(resourceReleaser); commitChronicleMapReady(map, raf, headerBuffer, headerSize); - establishReplication(map); return map; } private VanillaChronicleMap openWithExistingFile( - File file, RandomAccessFile raf, boolean recover, boolean overrideBuilderConfig) + File file, RandomAccessFile raf, ChronicleHashResourceReleaser resourceReleaser, + boolean recover, boolean overrideBuilderConfig) throws IOException { try { int headerSize = waitUntilReady(raf, recover); @@ -1645,16 +1656,15 @@ public final class ChronicleMapBuilder implements } map.initTransientsFromBuilder(this); if (!recover) { - map.createMappedStoreAndSegments(); + map.createMappedStoreAndSegments(resourceReleaser); } else { if (!overrideBuilderConfig) writeNotComplete(fileChannel, headerBuffer, headerSize); // if overrideBuilderConfig = true, readiness bit is already set // in writeHeader() call - map.recover(); + map.recover(resourceReleaser); commitChronicleMapReady(map, raf, headerBuffer, headerSize); } - establishReplication(map); return map; } catch (Exception e) { if (recover && !(e instanceof IOException) && @@ -1668,17 +1678,20 @@ public final class ChronicleMapBuilder implements replicated = replicationIdentifier != -1; persisted = false; + ChronicleHashResourceReleaser resourceReleaser = + new InMemoryChronicleHashResourceReleaser(); try { VanillaChronicleMap map = newMap(); - BytesStore bytesStore = - lazyNativeBytesStoreWithFixedCapacity(map.sizeInBytesWithoutTiers()); - map.createMappedStoreAndSegments(bytesStore); - establishReplication(map); + map.createInMemoryStoreAndSegments(resourceReleaser); prepareMapPublication(map); return map; - } catch (IOException e) { - // file-less version should never trigger an IOException. - throw new AssertionError(e); + } catch (Throwable t) { + try { + resourceReleaser.releaseManually(); + } catch (Exception e) { + t.addSuppressed(e); + } + throw Throwables.propagate(t); } } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index b635a65b2..002e23f71 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -150,8 +150,8 @@ public class VanillaChronicleMap initOwnTransients(); } - public void recover() throws IOException { - basicRecover(); + public void recover(ChronicleHashResourceReleaser resourceReleaser) throws IOException { + basicRecover(resourceReleaser); try (IterationContext iterationContext = iterationContext()) { iterationContext.recoverSegments(); } diff --git a/src/main/java/net/openhft/chronicle/set/SetFromMap.java b/src/main/java/net/openhft/chronicle/set/SetFromMap.java index 8772af45e..705527c26 100644 --- a/src/main/java/net/openhft/chronicle/set/SetFromMap.java +++ b/src/main/java/net/openhft/chronicle/set/SetFromMap.java @@ -116,6 +116,11 @@ class SetFromMap extends AbstractSet implements ChronicleSet { return m.longSize(); } + @Override + public long offHeapMemoryUsed() { + return m.offHeapMemoryUsed(); + } + @Override public Class keyClass() { return m.keyClass(); diff --git a/src/test/java/net/openhft/chronicle/map/MemoryLeakTest.java b/src/test/java/net/openhft/chronicle/map/MemoryLeakTest.java deleted file mode 100644 index f98a99bf4..000000000 --- a/src/test/java/net/openhft/chronicle/map/MemoryLeakTest.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * Copyright (C) 2012, 2016 higherfrequencytrading.com - * Copyright (C) 2016 Roman Leventov - * - * 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 org.junit.Assert; -import org.junit.Test; - -import java.lang.ref.WeakReference; - -public class MemoryLeakTest { - - @Test - public void testChronicleMapCollected() { - WeakReference> ref = getMapRef(false); - System.gc(); - Assert.assertNull(ref.get()); - } - - @Test - public void testReplicatedChronicleMapCollected() { - WeakReference> ref = getMapRef(true); - System.gc(); - Assert.assertNull(ref.get()); - } - - private static WeakReference> getMapRef(boolean replicated) { - ChronicleMapBuilder builder = ChronicleMap - .of(Integer.class, Integer.class); - if (replicated) - builder.replication((byte) 1); - ChronicleMap map = builder.entries(1).create(); - map.put(1, 1); - return new WeakReference<>(map); - } -} diff --git a/src/test/java/net/openhft/chronicle/map/MemoryResourcesTest.java b/src/test/java/net/openhft/chronicle/map/MemoryResourcesTest.java new file mode 100644 index 000000000..73f2d2c5d --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/MemoryResourcesTest.java @@ -0,0 +1,123 @@ +/* + * Copyright (C) 2012, 2016 higherfrequencytrading.com + * Copyright (C) 2016 Roman Leventov + * + * 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.bytes.NoBytesStore; +import net.openhft.chronicle.core.OS; +import net.openhft.chronicle.core.values.IntValue; +import net.openhft.chronicle.values.Values; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import java.io.IOException; +import java.lang.ref.WeakReference; +import java.util.Arrays; +import java.util.Collection; + +import static org.junit.Assert.assertEquals; + +@RunWith(Parameterized.class) +public class MemoryResourcesTest { + + @Parameterized.Parameters + public static Collection data() { + return Arrays.asList(new Object[][] { + { false, false }, { false, true }, { true, false }, { true, true } + }); + } + + @Rule + public TemporaryFolder folder = new TemporaryFolder(); + + private boolean replicated; + private boolean persisted; + + public MemoryResourcesTest(boolean replicated, boolean persisted) { + this.replicated = replicated; + this.persisted = persisted; + } + + @Before + public void initNoBytesStore() { + Assert.assertNotEquals(0, NoBytesStore.NO_PAGE); + } + + @Test(timeout = 60_000) + public void testChronicleMapCollectedAndDirectMemoryReleased() throws IOException { + long nativeMemoryUsedBeforeMap = nativeMemoryUsed(); + WeakReference> ref = new WeakReference<>(getMap()); + Assert.assertNotNull(ref.get()); + long expectedNativeMemory = nativeMemoryUsedBeforeMap + ref.get().offHeapMemoryUsed(); + assertEquals(expectedNativeMemory, nativeMemoryUsed()); + // Wait until Map is collected by GC + while (ref.get() != null) { + System.gc(); + Thread.yield(); + } + // Wait until Cleaner is called and memory is returned to the system + while (nativeMemoryUsedBeforeMap != nativeMemoryUsed()) { + System.gc(); + Thread.yield(); + } + } + + private long nativeMemoryUsed() { + if (persisted) { + return OS.memoryMapped(); + } else { + return OS.memory().nativeMemoryUsed(); + } + } + + @Test + public void testExplicitChronicleMapCloseReleasesMemory() throws IOException { + long nativeMemoryUsedBeforeMap = nativeMemoryUsed(); + try (ChronicleMap map = getMap()) { + long expectedNativeMemory = nativeMemoryUsedBeforeMap + map.offHeapMemoryUsed(); + assertEquals(expectedNativeMemory, nativeMemoryUsed()); + } + assertEquals(nativeMemoryUsedBeforeMap, nativeMemoryUsed()); + } + + private ChronicleMap getMap() throws IOException { + ChronicleMapBuilder builder = ChronicleMap + .of(IntValue.class, String.class); + if (replicated) + builder.replication((byte) 1); + builder.entries(1).averageValueSize(1); + VanillaChronicleMap map; + if (persisted) { + map = (VanillaChronicleMap) + builder.createPersistedTo(folder.newFile("map.dat")); + } else { + map = (VanillaChronicleMap) builder.create(); + } + IntValue key = Values.newHeapInstance(IntValue.class); + int i = 0; + while (!map.hasExtraTierBulks()) { + key.setValue(i++); + map.put(key, "string" + i); + } + return map; + } +} diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index f22daa307..c8a9a285f 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -214,6 +214,11 @@ public class ReplicationCheckingMap implements ChronicleMap { return map1.longSize(); } + @Override + public long offHeapMemoryUsed() { + return map1.offHeapMemoryUsed(); + } + @Override public V getUsing(K key, V usingValue) { return map1.getUsing(key, usingValue);