diff --git a/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java index ff001d31d..19683b287 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java @@ -52,7 +52,7 @@ public final class BigSegmentHeader implements SegmentHeader { * Previously this value was 2 seconds, but GC pauses often take more time that shouldn't * result to IllegalStateException. */ - private static final int LOCK_TIMEOUT_SECONDS = 60; + public static final int LOCK_TIMEOUT_SECONDS = 60; private static RuntimeException deadLock() { return new RuntimeException("Failed to acquire the lock in " + LOCK_TIMEOUT_SECONDS + 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 2440243f3..081d5cf75 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -23,6 +23,7 @@ import net.openhft.chronicle.bytes.NativeBytesStore; import net.openhft.chronicle.core.Maths; 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; @@ -44,10 +45,10 @@ import sun.misc.Cleaner; import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; +import java.lang.ref.WeakReference; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; -import java.util.ArrayList; -import java.util.List; +import java.util.*; import java.util.concurrent.TimeUnit; import static java.lang.Long.numberOfTrailingZeros; @@ -156,12 +157,14 @@ public abstract class VanillaChronicleHash> allContexts; + public VanillaChronicleHash(ChronicleMapBuilder builder) { createdOrInMemory = true; - closed = false; // Version dataFileVersion = BuildVersion.version(); @@ -227,12 +230,10 @@ public abstract class VanillaChronicleHash "dataFileVersion").text(); @@ -370,6 +371,8 @@ public abstract class VanillaChronicleHash(); if (tierHashLookupSlotSize == 4) { @@ -380,6 +383,7 @@ public abstract class VanillaChronicleHash(); } public final void initBeforeMapping( @@ -633,20 +637,41 @@ public abstract class VanillaChronicleHash contextRef : allContexts) { + ChainingInterface context = contextRef.get(); + if (context != null && context.owner().isAlive()) { + // Ensures that if the thread owning this context will come to access + // chronicleHash concurrently with resource releasing operations below, it will + // fail due to the check in context.lockContextLocally() method. + // If the thread owning this context is currently accessing chronicleHash, + // closeContext() will spin-wait until the end of this access session. + context.closeContext(); + } + } + + bs.release(); + assert bs.refCount() == 0; + for (TierBulkData bulkData : tierBulkOffsets) { + if (bulkData.bytesStore != bs) { + bulkData.bytesStore.release(); + assert bulkData.bytesStore.refCount() == 0; + } + } + if (rafCleaner != null) + rafCleaner.clean(); } - if (rafCleaner != null) - rafCleaner.clean(); } @Override @@ -984,4 +1009,42 @@ public abstract class VanillaChronicleHash(context)); + } + } + + /** For testing only */ + public List> allContexts() { + return Collections.unmodifiableList(allContexts); + } + + private static final Collection> pseudoCollectionForExpunge = + new AbstractCollection>() { + + @Override + public boolean contains(Object o) { + ChainingInterface context = ((WeakReference) o).get(); + return context == null || !context.owner().isAlive(); + } + + @Override + public Iterator> iterator() { + throw new UnsupportedOperationException(); + } + + @Override + public int size() { + throw new UnsupportedOperationException(); + } + }; } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java index e6851d00d..8fb38a987 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java @@ -259,6 +259,12 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { public abstract boolean locksInit(); void initLocks() { + // This is a dummy check needed to make the Locks stage depend on the chaining.Used stage, + // to move the chaining.Used stage up in topological order, to make it closed later in + // the global context close() method, to ensure the context is unlocked (this is done in + // the chaining.closeUsed() method) after all other stages, potentially accessing Map's + // off-heap memory, are closed. + assert chaining.used; // Ensure SegmentHeader is init. This method initLocks() doesn't trigger SegmentHeader // initialization otherwise, so it could remain uninit on the beginning of locking methods. // But Locks still anyway depends on SegmentHeader stage (at least via diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java index d209736df..772e36ba3 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.hash.impl.stage.hash; +import net.openhft.sg.Stage; import net.openhft.sg.Staged; import java.util.ArrayList; @@ -27,6 +28,9 @@ public abstract class Chaining extends ChainingInterface { public final List contextChain; public final int indexInContextChain; + /** + * First context, ever created in this thread. rootContextInThisThread === contextChain.get(0). + */ public final ChainingInterface rootContextInThisThread; public Chaining() { @@ -53,20 +57,30 @@ public abstract class Chaining extends ChainingInterface { return (T) contextChain.get(index); } - boolean used; + + @Stage("Used") public boolean used; + @Stage("Used") private boolean firstContextLockedInThisThread; @Override public boolean usedInit() { return used; } + /** + * Init method parameterized to avoid automatic initialization. {@code used} argument should + * always be {@code true}. + */ @Override public void initUsed(boolean used) { - this.used = used; + assert used; + firstContextLockedInThisThread = rootContextInThisThread.lockContextLocally(); + this.used = true; } void closeUsed() { used = false; + if (firstContextLockedInThisThread) + rootContextInThisThread.unlockContextLocally(); } @Override diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java index 8f5d9341f..80871edd1 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java @@ -16,7 +16,6 @@ package net.openhft.chronicle.hash.impl.stage.hash; -import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.hash.impl.stage.entry.ReadLock; import net.openhft.chronicle.hash.impl.stage.entry.UpdateLock; import net.openhft.chronicle.hash.impl.stage.entry.WriteLock; @@ -28,7 +27,6 @@ import net.openhft.sg.Staged; public abstract class CheckOnEachPublicOperation { @StageRef OwnerThreadHolder holder; - @StageRef VanillaChronicleHashHolder hh; public void checkOnEachPublicOperation() { checkOnEachLockOperation(); @@ -42,7 +40,5 @@ public abstract class CheckOnEachPublicOperation { */ public void checkOnEachLockOperation() { holder.checkAccessingFromOwnerThread(); - if (!hh.h().isOpen()) - throw new IllegalStateException("Access to Chronicle Hash after close()"); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java index 30da91ef0..3a08e5201 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java @@ -16,8 +16,114 @@ package net.openhft.chronicle.hash.impl.stage.hash; +import net.openhft.chronicle.core.Memory; +import net.openhft.chronicle.core.OS; +import net.openhft.chronicle.hash.ChronicleHash; + +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.concurrent.TimeUnit; + +import static net.openhft.chronicle.hash.impl.BigSegmentHeader.LOCK_TIMEOUT_SECONDS; + public abstract class ThreadLocalState { public boolean iterationContextLockedInThisThread; + private static final Memory MEMORY = OS.memory(); + private static final long CONTEXT_LOCK_OFFSET; + private static final int CONTEXT_UNLOCKED = 0; + private static final int CONTEXT_LOCKED_LOCALLY = 1; + private static final int CONTEXT_CLOSED = 2; + + static { + try { + Field contextLockField = + ThreadLocalState.class.getDeclaredField("contextLock"); + contextLockField.setAccessible(true); + CONTEXT_LOCK_OFFSET = MEMORY.getFieldOffset(contextLockField); + } catch (NoSuchFieldException | SecurityException e) { + throw new AssertionError(e); + } + } + + private volatile int contextLock = CONTEXT_UNLOCKED; + + /** + * Returns {@code true} if this is the outer context lock in this thread, {@code false} if this + * is a nested context. + */ + public boolean lockContextLocally() { + // h().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, + CONTEXT_UNLOCKED, CONTEXT_LOCKED_LOCALLY)) { + return true; + } else { + if (contextLock == CONTEXT_LOCKED_LOCALLY) + return false; + if (contextLock == CONTEXT_CLOSED) + throw new IllegalStateException("Access to ChronicleHash after close()"); + throw new AssertionError("Unknown context lock state: " + contextLock); + } + } + + public void unlockContextLocally() { + // Ensure all reads from mapped memory are done before thread calling chronicleMap.close() + // frees resources potentially nulling some pointer used in those reads. + MEMORY.loadFence(); + // Avoid volatile write to avoid expensive store-load barrier + MEMORY.writeOrderedInt(this, CONTEXT_LOCK_OFFSET, CONTEXT_UNLOCKED); + } + + public void closeContext() { + if (tryCloseContext()) + return; + if (owner() == Thread.currentThread()) + throw new IllegalStateException("Attempt to close a Chronicle Hash in the context " + + "of not yet finished query or iteration"); + // Double the current timeout for segment locks "without timeout", that effectively + // specifies maximum lock (hence context) holding time + long timeoutMillis = TimeUnit.SECONDS.toMillis(LOCK_TIMEOUT_SECONDS) * 2; + long lastTime = System.currentTimeMillis(); + do { + if (tryCloseContext()) + return; + Thread.yield(); + long now = System.currentTimeMillis(); + if (now != lastTime) { + lastTime = now; + timeoutMillis--; + } + } while (timeoutMillis >= 0); + throw new RuntimeException("Failed to close a context, belonging to the thread\n" + + owner() + ", in the state: " + owner().getState() + ", current stack trace:\n" + + Arrays.toString(owner().getStackTrace()) + "\n" + + "Possible reasons:\n" + + "- The context owner thread exited before closing this context. Ensure that you\n" + + "always close opened Chronicle Map's contexts, the best way to do this is to use\n" + + "try-with-resources blocks." + + "- The context owner thread runs some context operation (e. g. a query) for\n" + + "unexpectedly long time (at least " + LOCK_TIMEOUT_SECONDS + " seconds).\n" + + "You should either redesign your logic to spend less time in Chronicle Map\n" + + "contexts (recommended) or synchronize map.close() with queries externally,\n" + + "so that close() is called only after all query operations finished.\n" + + "- Iteration over a large Chronicle Map takes more than " + LOCK_TIMEOUT_SECONDS + + " seconds.\n" + + "In this case you should synchronize map.close() with iterations over the map\n" + + "externally, so that close() is called only after all iterations are finished.\n" + + "- This is a dead lock involving the context owner thread and this thread (from\n" + + "which map.close() method is called. Make sure you always close Chronicle Map\n" + + "contexts, preferably using try-with-resources blocks."); + } + + private boolean tryCloseContext() { + return MEMORY.compareAndSwapInt(this, CONTEXT_LOCK_OFFSET, + CONTEXT_UNLOCKED, CONTEXT_CLOSED); + } + public abstract Thread owner(); + + public abstract ChronicleHash hash(); } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 52b7a65d3..f56b431a0 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -563,6 +563,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap(ReplicatedChronicleMap.this); + addContext(queryContext); cxt.set(queryContext); } return queryContext; @@ -597,6 +598,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap(ReplicatedChronicleMap.this); + addContext(iterContext); cxt.set(iterContext); } return iterContext; diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 23ed24339..38c15f568 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -34,9 +34,7 @@ import net.openhft.chronicle.wire.WireIn; import net.openhft.chronicle.wire.WireOut; import org.jetbrains.annotations.NotNull; -import java.io.File; import java.io.IOException; -import java.io.RandomAccessFile; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Set; @@ -248,6 +246,7 @@ public class VanillaChronicleMap queryContext = cxt.get(); if (queryContext == null) { queryContext = new CompiledMapQueryContext<>(VanillaChronicleMap.this); + addContext(queryContext); cxt.set(queryContext); } return queryContext; @@ -263,6 +262,7 @@ public class VanillaChronicleMap iterContext = cxt.get(); if (iterContext == null) { iterContext = new CompiledMapIterationContext<>(VanillaChronicleMap.this); + addContext(iterContext); cxt.set(iterContext); } return iterContext; diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java index fa2820e43..ba4d9e36c 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java @@ -57,6 +57,7 @@ public abstract class ReplicatedChronicleMapHolderImpl extends Chaining return m.chronicleSet; } + @Override public ChronicleHash hash() { return set() != null ? set() : map(); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java index e2eda3b95..23deba40f 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java @@ -57,6 +57,7 @@ public abstract class VanillaChronicleMapHolderImpl extends Chaining return m.chronicleSet; } + @Override public ChronicleHash hash() { return set() != null ? set() : map(); } diff --git a/src/test/java/net/openhft/chronicle/map/MapCloseTest.java b/src/test/java/net/openhft/chronicle/map/MapCloseTest.java new file mode 100644 index 000000000..7cfd5ed2e --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/MapCloseTest.java @@ -0,0 +1,92 @@ +/* + * 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.impl.stage.hash.ChainingInterface; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Semaphore; + +public class MapCloseTest { + + @Test(expected = IllegalStateException.class) + public void closeInContextTest() { + ChronicleMap map = + ChronicleMap.of(Integer.class, Integer.class).entries(1).create(); + ExternalMapQueryContext cxt = map.queryContext(1); + map.close(); + } + + @Test(expected = RuntimeException.class) + public void closeWithContextInAnotherThreadTest() throws InterruptedException { + ChronicleMap map = + ChronicleMap.of(Integer.class, Integer.class).entries(1).create(); + Object lock = new Object(); + CountDownLatch latch = new CountDownLatch(1); + synchronized (lock) { + new Thread() { + @Override + public void run() { + ExternalMapQueryContext cxt = map.queryContext(1); + latch.countDown(); + synchronized (lock) { + cxt.close(); + } + } + }.start(); + latch.await(); + map.close(); + } + } + + @Test + public void vanillaChronicleHashAllContextsExpungeTest() throws InterruptedException { + VanillaChronicleMap map = + (VanillaChronicleMap) + ChronicleMap.of(Integer.class, Integer.class).entries(1).create(); + Semaphore semaphore = new Semaphore(0); + CountDownLatch latch = new CountDownLatch(2); + class MapAccessThread extends Thread { + @Override + public void run() { + map.get(1); + latch.countDown(); + try { + semaphore.acquire(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + } + MapAccessThread t1 = new MapAccessThread(); + MapAccessThread t2 = new MapAccessThread(); + t1.start(); + t2.start(); + latch.await(); + Assert.assertEquals(2, map.allContexts().size()); + semaphore.release(2); + t1.join(); + t2.join(); + + map.get(1); + Assert.assertEquals(1, map.allContexts().size()); + ChainingInterface cxt = map.allContexts().get(0).get(); + Assert.assertTrue(cxt == map.queryContext(1)); + } +} diff --git a/src/test/java/net/openhft/chronicle/map/NestedContextsInIterationContextTest.java b/src/test/java/net/openhft/chronicle/map/NestedContextsInIterationContextTest.java index 81519375e..b770f0ee9 100644 --- a/src/test/java/net/openhft/chronicle/map/NestedContextsInIterationContextTest.java +++ b/src/test/java/net/openhft/chronicle/map/NestedContextsInIterationContextTest.java @@ -52,7 +52,8 @@ public class NestedContextsInIterationContextTest { .actualSegments(2) .create()) { map.put(42, 42); - int keySegmentIndex = map.queryContext(42).segmentIndex(); + ExternalMapQueryContext queryCxt = map.queryContext(42); + int keySegmentIndex = queryCxt.segmentIndex(); try (MapSegmentContext cxt = map.segmentContext(keySegmentIndex)) { cxt.forEachSegmentEntry(e -> { int k = 0; @@ -80,6 +81,8 @@ public class NestedContextsInIterationContextTest { } throw new AssertionError("expected " + IllegalStateException.class); }); + } finally { + queryCxt.close(); } } } @@ -92,7 +95,8 @@ public class NestedContextsInIterationContextTest { .actualSegments(2) .create()) { map.put(42, 42); - int keySegmentIndex = map.queryContext(42).segmentIndex(); + ExternalMapQueryContext queryCxt = map.queryContext(42); + int keySegmentIndex = queryCxt.segmentIndex(); try (MapSegmentContext cxt = map.segmentContext(keySegmentIndex)) { cxt.forEachSegmentEntry(e -> { int k = 0; @@ -120,6 +124,8 @@ public class NestedContextsInIterationContextTest { } throw new AssertionError("expected " + IllegalStateException.class); }); + } finally { + queryCxt.close(); } } } @@ -132,7 +138,8 @@ public class NestedContextsInIterationContextTest { .actualSegments(2) .create()) { map.put(42, 42); - int keySegmentIndex = map.queryContext(42).segmentIndex(); + ExternalMapQueryContext queryCxt = map.queryContext(42); + int keySegmentIndex = queryCxt.segmentIndex(); try (MapSegmentContext cxt = map.segmentContext(keySegmentIndex)) { cxt.forEachSegmentEntry(e -> { int k = 0; @@ -154,6 +161,8 @@ public class NestedContextsInIterationContextTest { nestedC.readLock().lock(); } }); + } finally { + queryCxt.close(); } } }