Prevent leaks of ChronicleMaps though thread local contexts and supplimental OldDeletedEntriesCleanupThread, add MemoryLeakTest

This commit is contained in:
leventov
2016-10-01 20:33:16 +03:00
parent 8981ae2fd0
commit 223b2d4329
16 changed files with 280 additions and 79 deletions
@@ -92,6 +92,11 @@ public abstract class CompactOffHeapLinearHashTable {
this.valueMask = mask(valueBits);
}
/**
* Must not store {@code h} in a field, to avoid memory leaks.
*
* @see net.openhft.chronicle.hash.impl.stage.hash.Chaining#initMap
*/
CompactOffHeapLinearHashTable(VanillaChronicleHash h) {
this(h.tierHashLookupCapacity, h.tierHashLookupSlotSize, h.tierHashLookupKeyBits,
h.tierHashLookupValueBits);
@@ -23,6 +23,11 @@ public final class IntCompactOffHeapLinearHashTable extends CompactOffHeapLinear
private static final long SCALE = 4L;
/**
* Must not store {@code h} in a field, to avoid memory leaks.
*
* @see net.openhft.chronicle.hash.impl.stage.hash.Chaining#initMap
*/
IntCompactOffHeapLinearHashTable(VanillaChronicleHash h) {
super(h);
}
@@ -23,6 +23,11 @@ public final class LongCompactOffHeapLinearHashTable extends CompactOffHeapLinea
private static final long SCALE = 8L;
/**
* Must not store {@code h} in a field, to avoid memory leaks.
*
* @see net.openhft.chronicle.hash.impl.stage.hash.Chaining#initMap
*/
LongCompactOffHeapLinearHashTable(VanillaChronicleHash h) {
super(h);
}
@@ -17,6 +17,9 @@
package net.openhft.chronicle.hash.impl;
import net.openhft.sg.Staged;
@Staged
public interface VanillaChronicleHashHolder<K> {
VanillaChronicleHash<K, ?, ?, ?> h();
}
@@ -17,12 +17,13 @@
package net.openhft.chronicle.hash.impl.stage.hash;
import net.openhft.chronicle.map.VanillaChronicleMap;
import net.openhft.sg.Stage;
import net.openhft.sg.Staged;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import java.util.function.BiFunction;
@Staged
public abstract class Chaining extends ChainingInterface {
@@ -34,18 +35,20 @@ public abstract class Chaining extends ChainingInterface {
*/
public final ChainingInterface rootContextInThisThread;
public Chaining() {
public Chaining(VanillaChronicleMap map) {
contextChain = new ArrayList<>();
contextChain.add(this);
indexInContextChain = 0;
rootContextInThisThread = this;
initMap(map);
}
public Chaining(ChainingInterface rootContextInThisThread) {
public Chaining(ChainingInterface rootContextInThisThread, VanillaChronicleMap map) {
contextChain = rootContextInThisThread.getContextChain();
indexInContextChain = contextChain.size();
contextChain.add(this);
this.rootContextInThisThread = rootContextInThisThread;
initMap(map);
}
@Override
@@ -58,6 +61,27 @@ public abstract class Chaining extends ChainingInterface {
return (T) contextChain.get(index);
}
/**
* This method stores a reference to the context's owner ChronicleMap into a field of the
* context in the beginning of each usage of the context. Previously, this field was final and
* set only once during context creation. It was preventing ChronicleMap objects from becoming
* dead and collected by the GC, while any thread, from which the ChronicleMap was accessed
* (i. e, a thread local context created), is alive.
*
* <p>The chain of strong references:
* 1) Thread ->
* 2) ThreadLocalMap ->
* 3) Entry with ThreadLocal {@link net.openhft.chronicle.map.VanillaChronicleMap#cxt} as weak
* referent and a context (e. g. {@link net.openhft.chronicle.map.impl.CompiledMapQueryContext})
* as value (a simple field, not a weak reference!) ->
* 4) final reference to the owner {@link VanillaChronicleMap} ->
* 5) ThreadLocal {@link net.openhft.chronicle.map.VanillaChronicleMap#cxt} (a strong reference
* this time! note that this ThreadLocal is an instance field of VanillaChronicleMap)
*
* <p>So in order to break this chain at step 4), contexts store references to their owner
* ChronicleMaps only when contexts are used.
*/
public abstract void initMap(VanillaChronicleMap map);
@Stage("Used") public boolean used;
@Stage("Used") private boolean firstContextLockedInThisThread;
@@ -72,12 +96,14 @@ public abstract class Chaining extends ChainingInterface {
* always be {@code true}.
*/
@Override
public void initUsed(boolean used) {
public void initUsed(boolean used, VanillaChronicleMap map) {
assert used;
firstContextLockedInThisThread = rootContextInThisThread.lockContextLocally();
initMap(map);
firstContextLockedInThisThread = rootContextInThisThread.lockContextLocally(map);
this.used = true;
}
@SuppressWarnings("unused")
void closeUsed() {
used = false;
if (firstContextLockedInThisThread)
@@ -86,10 +112,11 @@ public abstract class Chaining extends ChainingInterface {
@Override
public <T extends ChainingInterface> T getContext(
Class<? extends T> contextClass, Function<ChainingInterface, T> createChaining) {
Class<? extends T> contextClass, BiFunction<ChainingInterface, VanillaChronicleMap, T> createChaining,
VanillaChronicleMap map) {
for (ChainingInterface context : contextChain) {
if (context.getClass() == contextClass && !context.usedInit()) {
context.initUsed(true);
context.initUsed(true, map);
//noinspection unchecked
return (T) context;
}
@@ -104,8 +131,8 @@ public abstract class Chaining extends ChainingInterface {
"stack trace on https://github.com/OpenHFT/Chronicle-Map/issues");
}
//noinspection unchecked
T context = createChaining.apply(this);
context.initUsed(true);
T context = createChaining.apply(this, map);
context.initUsed(true, map);
return context;
}
}
@@ -17,17 +17,20 @@
package net.openhft.chronicle.hash.impl.stage.hash;
import net.openhft.chronicle.map.VanillaChronicleMap;
import java.util.List;
import java.util.function.Function;
import java.util.function.BiFunction;
public abstract class ChainingInterface extends ThreadLocalState {
public abstract List<ChainingInterface> getContextChain();
public abstract void initUsed(boolean used);
public abstract void initUsed(boolean used, VanillaChronicleMap map);
public abstract boolean usedInit();
public abstract <T extends ChainingInterface> T getContext(
Class<? extends T> contextClass, Function<ChainingInterface, T> createChaining);
Class<? extends T> contextClass, BiFunction<ChainingInterface, VanillaChronicleMap, T> createChaining,
VanillaChronicleMap map);
}
@@ -54,12 +54,12 @@ public abstract class ThreadLocalState {
* Returns {@code true} if this is the outer context lock in this thread, {@code false} if this
* is a nested context.
*/
public boolean lockContextLocally() {
public boolean lockContextLocally(ChronicleHash<?, ?, ?, ?> hash) {
// hash().isOpen() check guarantees no starvation of a thread calling chMap.close() and
// trying to close this context by closeContext() method below, while the thread owning this
// context frequently locks and unlocks it (e. g. in a loop). This is also the only check
// for chMap openness during the whole context usage lifecycle.
if (hash().isOpen() && MEMORY.compareAndSwapInt(this, CONTEXT_LOCK_OFFSET,
if (hash.isOpen() && MEMORY.compareAndSwapInt(this, CONTEXT_LOCK_OFFSET,
CONTEXT_UNLOCKED, CONTEXT_LOCKED_LOCALLY)) {
return true;
} else {
@@ -67,7 +67,7 @@ public abstract class ThreadLocalState {
return false;
// Don't extract this hash().isOpen() and the one above, because they could different
// results (the first (above) could return true, the second (below) - false).
if (!hash().isOpen())
if (!hash.isOpen())
throw new ChronicleHashClosedException();
throw new AssertionError("Unknown context lock state: " + contextLock);
}
@@ -134,6 +134,4 @@ public abstract class ThreadLocalState {
}
public abstract Thread owner();
public abstract ChronicleHash<?, ?, ?, ?> hash();
}
@@ -18,11 +18,13 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.core.io.Closeable;
import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
import net.openhft.chronicle.hash.ReplicatedHashSegmentContext;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.ref.WeakReference;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
@@ -36,7 +38,27 @@ class OldDeletedEntriesCleanupThread extends Thread
implements Closeable, Predicate<ReplicableEntry> {
private static final Logger LOG = LoggerFactory.getLogger(OldDeletedEntriesCleanupThread.class);
private final ReplicatedChronicleMap<?, ?, ?> map;
/**
* Don't store a strong ref to a map in order to avoid it's leaking, if the user forgets to
* close() map, from where this thread is shut down explicitly. Dereference map within
* a single method, {@link #cleanupSegment()}. The map has a chance to be collected by GC when
* this thread is sleeping after cleaning up a segment.
*/
private final WeakReference<ReplicatedChronicleMap<?, ?, ?>> mapRef;
/**
* {@code cleanupTimeout}, {@link #cleanupTimeoutUnit} and {@link #segments} are parts of the
* cleaned Map's state, extracted in order to minimize accesses to the map.
*
* @see ChronicleHashBuilderPrivateAPI#removedEntryCleanupTimeout(long, TimeUnit)
*/
private final long cleanupTimeout;
private final TimeUnit cleanupTimeoutUnit;
private final int segments;
/**
* {@code segmentsPermutation} and {@link #inverseSegmentsPermutation} determine random order,
* in which segments are cleaned up.
*/
private final int[] segmentsPermutation;
private final int[] inverseSegmentsPermutation;
@@ -52,9 +74,13 @@ class OldDeletedEntriesCleanupThread extends Thread
private long prevSegment0ScanStart = -1;
private long removedCompletely;
public OldDeletedEntriesCleanupThread(ReplicatedChronicleMap<?, ?, ?> map) {
OldDeletedEntriesCleanupThread(ReplicatedChronicleMap<?, ?, ?> map) {
super("Cleanup Thread for ReplicatedChronicleMap persisted to " + map.file());
this.map = map;
this.mapRef = new WeakReference<>(map);
cleanupTimeout = map.cleanupTimeout;
cleanupTimeoutUnit = map.cleanupTimeoutUnit;
segments = map.segments();
segmentsPermutation = randomPermutation(map.segments());
inverseSegmentsPermutation = inversePermutation(segmentsPermutation);
}
@@ -62,31 +88,14 @@ class OldDeletedEntriesCleanupThread extends Thread
@Override
public void run() {
while (!shutdown) {
int segmentIndex = map.globalMutableState().getCurrentCleanupSegmentIndex();
int nextSegmentIndex;
try (MapSegmentContext<?, ?, ?> context = map.segmentContext(segmentIndex)) {
if (segmentIndex == 0)
prevSegment0ScanStart = 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;
}
}
int nextSegmentIndex = cleanupSegment();
if (nextSegmentIndex == -1)
return;
if (nextSegmentIndex == 0) {
long currentTime = currentTime();
TimeUnit cleanupTimeoutUnit = map.cleanupTimeoutUnit;
long mapScanTime = systemTimeIntervalBetween(
prevSegment0ScanStart, currentTime, cleanupTimeoutUnit);
LOG.debug("Old deleted entries scan time: {} {}", mapScanTime, cleanupTimeoutUnit);
long cleanupTimeout = map.cleanupTimeout;
if (mapScanTime < cleanupTimeout) {
long timeToSleep = cleanupTimeoutUnit.toMillis(cleanupTimeout - mapScanTime);
if (timeToSleep > 0) {
@@ -99,14 +108,42 @@ class OldDeletedEntriesCleanupThread extends Thread
}
}
/**
* @return next segment index to cleanup, or -1 if cleanup thread should be shut down
*/
private int cleanupSegment() {
ReplicatedChronicleMap<?, ?, ?> map = mapRef.get();
if (map == null)
return -1;
int segmentIndex = map.globalMutableState().getCurrentCleanupSegmentIndex();
int nextSegmentIndex;
try (MapSegmentContext<?, ?, ?> context = map.segmentContext(segmentIndex)) {
if (segmentIndex == 0)
prevSegment0ScanStart = 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);
return nextSegmentIndex;
} else {
// forEachWhile returned false => interrupted => shutdown = true
assert shutdown;
return -1;
}
}
}
@Override
public boolean test(ReplicableEntry e) {
if (shutdown)
return false;
if (e instanceof MapAbsentEntry) {
long deleteTimeout = systemTimeIntervalBetween(
e.originTimestamp(), currentTime(), map.cleanupTimeoutUnit);
if (deleteTimeout > map.cleanupTimeout && !e.isChanged()) {
e.originTimestamp(), currentTime(), cleanupTimeoutUnit);
if (deleteTimeout > cleanupTimeout && !e.isChanged()) {
e.doRemoveCompletely();
removedCompletely++;
}
@@ -136,7 +173,7 @@ class OldDeletedEntriesCleanupThread extends Thread
private int nextSegmentIndex(int segmentIndex) {
int permutationIndex = inverseSegmentsPermutation[segmentIndex];
int nextPermutationIndex = (permutationIndex + 1) % map.segments();
int nextPermutationIndex = (permutationIndex + 1) % segments;
return segmentsPermutation[nextPermutationIndex];
}
@@ -572,7 +572,10 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
public CompiledReplicatedMapQueryContext<K, V, R> mapContext() {
//noinspection unchecked
return q().getContext(CompiledReplicatedMapQueryContext.class,
ci -> new CompiledReplicatedMapQueryContext<>(ci, this));
// lambda is used instead of constructor reference because currently stage-compiler
// has issues with parsing method/constructor refs.
// TODO replace with constructor ref when stage-compiler is improved
(c, m) -> new CompiledReplicatedMapQueryContext<K, V, R>(c, m), this);
}
/**
@@ -606,7 +609,10 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
public CompiledReplicatedMapIterationContext<K, V, R> iterationContext() {
//noinspection unchecked
return i().getContext(CompiledReplicatedMapIterationContext.class,
ci -> new CompiledReplicatedMapIterationContext<>(ci, this));
// lambda is used instead of constructor reference because currently stage-compiler
// has issues with parsing method/constructor refs.
// TODO replace with constructor ref when stage-compiler is improved
(c, m) -> new CompiledReplicatedMapIterationContext<K, V, R>(c, m), this);
}
/**
@@ -276,7 +276,10 @@ public class VanillaChronicleMap<K, V, R>
public QueryContextInterface<K, V, R> mapContext() {
//noinspection unchecked
return q().getContext(CompiledMapQueryContext.class,
ci -> new CompiledMapQueryContext<>(ci, this));
// lambda is used instead of constructor reference because currently stage-compiler
// has issues with parsing method/constructor refs.
// TODO replace with constructor ref when stage-compiler is improved
(c, m) -> new CompiledMapQueryContext<K, V, R>(c, m), this);
}
private ChainingInterface i() {
@@ -290,9 +293,14 @@ public class VanillaChronicleMap<K, V, R>
return iterContext;
}
public IterationContext<K, V, ?> iterationContext() {
return i().getContext(CompiledMapIterationContext.class,
ci -> new CompiledMapIterationContext<>(ci, this));
public IterationContext<K, V, R> iterationContext() {
//noinspection unchecked
return i().getContext(
CompiledMapIterationContext.class,
// lambda is used instead of constructor reference because currently stage-compiler
// has issues with parsing method/constructor refs.
// TODO replace with constructor ref when stage-compiler is improved
(c, m) -> new CompiledMapIterationContext<K, V, R>(c, m), this);
}
@Override
@@ -18,7 +18,9 @@
package net.openhft.chronicle.map.impl;
import net.openhft.chronicle.map.ReplicatedChronicleMap;
import net.openhft.sg.Staged;
@Staged
public interface ReplicatedChronicleMapHolder<K, V, R> extends VanillaChronicleMapHolder<K, V, R> {
@Override
@@ -22,19 +22,17 @@ import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.VanillaChronicleMap;
import net.openhft.chronicle.set.ChronicleSet;
import net.openhft.sg.Staged;
@Staged
public interface VanillaChronicleMapHolder<K, V, R> extends VanillaChronicleHashHolder<K> {
VanillaChronicleMap<K, V, R> m();
@Override
default VanillaChronicleHash<K, ?, ?, ?> h() {
return m();
}
VanillaChronicleHash<K, ?, ?, ?> h();
default ChronicleMap<K, V> map() {
return m();
}
ChronicleMap<K, V> map();
ChronicleSet<K> set();
}
@@ -18,29 +18,41 @@
package net.openhft.chronicle.map.impl.stage.map;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
import net.openhft.chronicle.hash.impl.stage.hash.Chaining;
import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ReplicatedChronicleMap;
import net.openhft.chronicle.map.VanillaChronicleMap;
import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder;
import net.openhft.chronicle.set.ChronicleSet;
import net.openhft.sg.Stage;
import net.openhft.sg.Staged;
@Staged
public abstract class ReplicatedChronicleMapHolderImpl<K, V, R> extends Chaining
public abstract class ReplicatedChronicleMapHolderImpl<K, V, R>
extends Chaining
implements ReplicatedChronicleMapHolder<K, V, R> {
private final ReplicatedChronicleMap<K, V, R> m;
@Stage("Map") private ReplicatedChronicleMap<K, V, R> m = null;
public ReplicatedChronicleMapHolderImpl(ReplicatedChronicleMap<K, V, R> m) {
super();
this.m = m;
public ReplicatedChronicleMapHolderImpl(VanillaChronicleMap map) {
super(map);
}
public ReplicatedChronicleMapHolderImpl(
ChainingInterface c, ReplicatedChronicleMap<K, V, R> m) {
super(c);
this.m = m;
ChainingInterface rootContextInThisThread, VanillaChronicleMap map) {
super(rootContextInThisThread, map);
}
@Override
public void initMap(VanillaChronicleMap map) {
// alternative to this "unsafe" casting approach is proper generalization
// of Chaining/ChainingInterface, but this causes issues with current version
// of stage-compiler.
// TODO generalize Chaining with <M extends VanillaCM> when stage-compiler is improved.
//noinspection unchecked
m = (ReplicatedChronicleMap<K, V, R>) map;
}
@Override
@@ -48,9 +60,14 @@ public abstract class ReplicatedChronicleMapHolderImpl<K, V, R> extends Chaining
return m;
}
@Override
public VanillaChronicleHash<K, ?, ?, ?> h() {
return m;
}
@Override
public ChronicleMap<K, V> map() {
return m();
return m;
}
@Override
@@ -58,7 +75,6 @@ public abstract class ReplicatedChronicleMapHolderImpl<K, V, R> extends Chaining
return m.chronicleSet;
}
@Override
public ChronicleHash<K, ?, ?, ?> hash() {
return set() != null ? set() : map();
}
@@ -18,29 +18,40 @@
package net.openhft.chronicle.map.impl.stage.map;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
import net.openhft.chronicle.hash.impl.stage.hash.Chaining;
import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.VanillaChronicleMap;
import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder;
import net.openhft.chronicle.set.ChronicleSet;
import net.openhft.sg.Stage;
import net.openhft.sg.Staged;
@Staged
public abstract class VanillaChronicleMapHolderImpl<K, V, R> extends Chaining
public abstract class VanillaChronicleMapHolderImpl<K, V, R>
extends Chaining
implements VanillaChronicleMapHolder<K, V, R> {
private final VanillaChronicleMap<K, V, R> m;
public VanillaChronicleMapHolderImpl(VanillaChronicleMap<K, V, R> m) {
super();
this.m = m;
@Stage("Map") private VanillaChronicleMap<K, V, R> m = null;
public VanillaChronicleMapHolderImpl(VanillaChronicleMap map) {
super(map);
}
public VanillaChronicleMapHolderImpl(
ChainingInterface c, VanillaChronicleMap<K, V, R> m) {
super(c);
this.m = m;
ChainingInterface rootContextInThisThread, VanillaChronicleMap map) {
super(rootContextInThisThread, map);
}
@Override
public void initMap(VanillaChronicleMap map) {
// alternative to this "unsafe" casting approach is proper generalization
// of Chaining/ChainingInterface, but this causes issues with current version
// of stage-compiler.
// TODO generalize Chaining with <M extends VanillaCM> when stage-compiler is improved.
//noinspection unchecked
m = map;
}
@Override
@@ -48,9 +59,14 @@ public abstract class VanillaChronicleMapHolderImpl<K, V, R> extends Chaining
return m;
}
@Override
public VanillaChronicleHash<K, ?, ?, ?> h() {
return m;
}
@Override
public ChronicleMap<K, V> map() {
return m();
return m;
}
@Override
@@ -58,7 +74,6 @@ public abstract class VanillaChronicleMapHolderImpl<K, V, R> extends Chaining
return m.chronicleSet;
}
@Override
public ChronicleHash<K, ?, ?, ?> hash() {
return set() != null ? set() : map();
}