diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashClosedException.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashClosedException.java index cdcf7b74b..86665719b 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashClosedException.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashClosedException.java @@ -24,6 +24,10 @@ public final class ChronicleHashClosedException extends IllegalStateException { private static final long serialVersionUID = 0L; public ChronicleHashClosedException(ChronicleHash hash) { - super("Access to " + hash.toIdentityString() + " after close()"); + this(hash.toIdentityString()); + } + + public ChronicleHashClosedException(String chronicleHashIdentityString) { + super("Access to " + chronicleHashIdentityString + " after close()"); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResources.java b/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResources.java index b89cbd713..85a677cc0 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResources.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashResources.java @@ -17,26 +17,48 @@ package net.openhft.chronicle.hash.impl; +import net.openhft.chronicle.hash.ChronicleHashClosedException; +import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.hash.impl.util.Throwables; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -import java.util.ArrayList; -import java.util.List; +import java.io.Closeable; +import java.lang.ref.WeakReference; +import java.util.*; /** * ChronicleHashResources is Runnable to be passed as "hunk" to {@link sun.misc.Cleaner}. */ public abstract class ChronicleHashResources implements Runnable { + private static final Logger LOG = LoggerFactory.getLogger(ChronicleHashResources.class); - List memoryResources = new ArrayList<>(1); + /** + * closed field is volatile and {@link #closed()} accessor is not synchronized, because it + * is called on each ChronicleMap access from all threads, synchronization would make closed() + * a bottleneck. + */ + private volatile boolean closed = false; + private List memoryResources = new ArrayList<>(); + private List closeables = new ArrayList<>(1); + private ArrayList> contexts = new ArrayList<>(); + /** + * Identity String of the ChronicleHash, for which this ChronicleHashResources is created. + * ChronicleHash couldn't be directly referenced, because it's a hunk for {@link + * sun.misc.Cleaner}, and it would prevent the chronicleHash from ever becoming unreachable. + */ + private String chronicleHashIdentityString; - public final synchronized void addMemoryResource(long address, long size) { - if (memoryResources == null) - throw new IllegalStateException("Already released"); - memoryResources.add(new MemoryResource(address, size)); + List memoryResources() { + return memoryResources; } - public final synchronized long totalMemory() { - if (memoryResources == null) + List> contexts() { + return contexts; + } + + final synchronized long totalMemory() { + if (closed) return 0L; long totalMemory = 0L; for (MemoryResource memoryResource : memoryResources) { @@ -45,18 +67,183 @@ public abstract class ChronicleHashResources implements Runnable { return totalMemory; } - abstract Throwable doRelease(); + final boolean closed() { + return closed; + } - public final synchronized void releaseManually() { - if (memoryResources == null) - return; // Already released + private void checkOpen() { + if (closed) + throw new ChronicleHashClosedException(chronicleHashIdentityString); + } + + public final synchronized void setChronicleHashIdentityString( + String chronicleHashIdentityString) { + this.chronicleHashIdentityString = chronicleHashIdentityString; + } + + final synchronized void addMemoryResource(long address, long size) { + checkOpen(); + memoryResources.add(new MemoryResource(address, size)); + } + + public final synchronized void addCloseable(Closeable closeable) { + checkOpen(); + closeables.add(closeable); + } + + final synchronized void addContext(ContextHolder contextHolder) { + checkOpen(); + expungeStateContexts(); + contexts.add(new WeakReference<>(contextHolder)); + } + + private void expungeStateContexts() { + contexts.removeIf(ref -> { + ContextHolder contextHolder = ref.get(); + return contextHolder == null || !contextHolder.get().owner().isAlive(); + }); + } + + /** + * This run() method is called from {@link sun.misc.Cleaner}, hence must not fail + */ + @Override + public void run() { Throwable thrown = null; try { - thrown = doRelease(); + if (closed) + return; + try { + LOG.error("{} not closed manually, cleaned up from Cleaner", + chronicleHashIdentityString); + } catch (Throwable t) { + thrown = t; + } finally { + synchronized (this) { + if (closed) { + LOG.error("Somebody closed {} while it is processed by Cleaner, " + + "this should be impossible", chronicleHashIdentityString); + } else { + thrown = Throwables.returnOrSuppress(thrown, releaseEverything()); + } + } + if (thrown != null) { + try { + LOG.error("Error on releasing resources of " + chronicleHashIdentityString, + thrown); + } catch (Throwable t) { + // This may occur if we are in shutdown hooks, and the log service has + // already been shut down. Try to fall back to printStackTrace(). + thrown.addSuppressed(t); + thrown.printStackTrace(); + } + } + } + } catch (Throwable ignore) { + // just don't fail anyway + } + } + + abstract Throwable releaseSystemResources(); + + public final boolean releaseManually() { + if (closed) + return false; + synchronized (this) { + if (closed) + return false; + Throwable thrown = releaseEverything(); + if (thrown != null) + throw Throwables.propagate(thrown); + return true; + } + } + + private Throwable releaseEverything() { + // It's important to set closed = true before calling closeContexts(), to ensure that + // threads which access this chronicleHash for the first time concurrently with + // chronicleHash.close() will either fail to register the context via addContext() (because + // checkOpen() is called there), or either the registered contexts will be visible and + // closed in closeContexts() + closed = true; + + // Paranoiac mode: methods closeContexts(), closeCloseables(), releaseSystemResources() and + // Throwables.returnOrSuppress() should never throw any throwables (the first three should + // return them instead of throwing), but we don't trust ourselves and wrap every call into + // it's own try-finally block + try { + Throwable thrown = null; + try { + thrown = closeContexts(); + } finally { + try { + thrown = Throwables.returnOrSuppress(thrown, closeCloseables()); + } finally { + // It's important to close the system resources after contexts and closeables + // because the latter may use those system resources + thrown = Throwables.returnOrSuppress(thrown, releaseSystemResources()); + } + } + return thrown; } finally { + // Make GC life easier + closeables = null; + contexts = null; memoryResources = null; } - if (thrown != null) - throw Throwables.propagate(thrown); + } + + private Throwable closeContexts() { + Throwable thrown = null; + // Indexed loop instead of for-each because we may be in the context of cleaning up after + // OutOfMemoryError, and allocating Iterator object may lead to another OutOfMemoryError, so + // we try to avoid any allocations + //noinspection ForLoopReplaceableByForEach + for (int i = 0; i < contexts.size(); i++) { + WeakReference contextHolderRef = contexts.get(i); + try { + ContextHolder contextHolder = contextHolderRef.get(); + if (contextHolder != null) { + closeContext(contextHolder); + } + } catch (Throwable t) { + thrown = Throwables.returnOrSuppress(thrown, t); + } + } + return thrown; + } + + private void closeContext(ContextHolder contextHolder) { + ChainingInterface context = contextHolder.get(); + try { + if (context.owner().isAlive()) { + // Ensures that if the thread owning this context will come to access chronicleHash + // concurrently with resource releasing operation, 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(chronicleHashIdentityString); + } + } finally { + // See ContextHolder's class-level Javadoc comment which explains motivation for this + contextHolder.clear(); + } + } + + private Throwable closeCloseables() { + Throwable thrown = null; + // Indexed loop instead of for-each because we may be in the context of cleaning up after + // OutOfMemoryError, and allocating Iterator object may lead to another OutOfMemoryError, so + // we try to avoid any allocations + //noinspection ForLoopReplaceableByForEach + for (int i = 0; i < closeables.size(); i++) { + Closeable closeable = closeables.get(i); + try { + closeable.close(); + } catch (Throwable t) { + thrown = Throwables.returnOrSuppress(thrown, t); + } + } + return thrown; } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/ContextHolder.java b/src/main/java/net/openhft/chronicle/hash/impl/ContextHolder.java new file mode 100644 index 000000000..522de85af --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/ContextHolder.java @@ -0,0 +1,69 @@ +/* + * 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.stage.hash.ChainingInterface; + +/** + * A simple wrapper of {@link ChainingInterface}, the ChainingInterface field could be set to null. + * + *

Motivation

+ *

{@link net.openhft.chronicle.map.ChronicleMap}'s context objects are huge and reference their + * own instances of key and value marshallers, which usually have buffers for serialization (e. g. + * see {@link net.openhft.chronicle.hash.serialization.impl.SerializableDataAccess}). The contexts + * are stored in {@link ThreadLocal}s, which are instance fields of ChronicleMap objects + * (see {@link net.openhft.chronicle.map.VanillaChronicleMap#cxt}). We want the context objects to + * be eligible for garbage collection as soon as possible after the ChronicleMap object is closed + * or becomes unreachable. + * + *

In JDK 8 ThreadLocals are implemented using {@link java.lang.ThreadLocal.ThreadLocalMap}, + * a hash table with ThreadLocal objects themselves as the keys, weak-referenced. So after + * ChronicleMap (hence it's ThreadLocal cxt field) becomes unreachable, context objects should be + * eventually removed from ThreadLocalMap and become unreachable, but the current implementation + * does some cleanup of ThreadLocalMap lazily and only on ThreadLocal.set(), initialValue() and + * remove(), but not on the hot path of ThreadLocal.get(). I. e. if there is not enough "ThreadLocal + * activity" within a thread, stale ChronicleMap contexts may not be removed from ThreadLocalMaps + * forever, effectively this is a memory leak. + * + *

Moreover, if the user of the library closes ChronicleMap with close(), but has ChronicleMap + * object leaked, ChronicleMap's ThreadLocal field doesn't become unreachable and the leak of + * context objects is "legitimate". + * + *

Solution for this is to reference from {@link + * net.openhft.chronicle.map.VanillaChronicleMap#cxt} not huge context object directly, but small + * ContextHolder object, and clear the reference to context via {@link #clear()} on + * ChronicleMap.close() or from {@link sun.misc.Cleaner}'s registered cleaner for ChronicleMap, if + * it weren't closed, but becomes unreachable and reclaimed by the garbage collector. + * + * @see ChronicleHashResources#closeContext(ContextHolder) + */ +public final class ContextHolder { + + private ChainingInterface context; + + public ContextHolder(ChainingInterface context) { + this.context = context; + } + + public ChainingInterface get() { + return context; + } + + void clear() { + context = null; + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResources.java b/src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResources.java index 886de64e1..9f60feae0 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResources.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/InMemoryChronicleHashResources.java @@ -18,51 +18,28 @@ package net.openhft.chronicle.hash.impl; import net.openhft.chronicle.core.OS; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import net.openhft.chronicle.hash.impl.util.Throwables; + +import java.util.List; public final class InMemoryChronicleHashResources extends ChronicleHashResources { - private static final Logger LOG = - LoggerFactory.getLogger(PersistedChronicleHashResources.class); - public InMemoryChronicleHashResources() {} @Override - public synchronized void run() { - if (memoryResources == null) - return; // Already released - // All Throwables are caught because this class is used as Runnable for sun.misc.Cleaner, - // hence must not fail - try { - Throwable thrown = doRelease(); - 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(); - } - } - } finally { - memoryResources = null; - } - } - - @Override - Throwable doRelease() { + Throwable releaseSystemResources() { Throwable thrown = null; - for (MemoryResource allocation : memoryResources) { + List memoryResources = memoryResources(); + // Indexed loop instead of for-each because we may be in the context of cleaning up after + // OutOfMemoryError, and allocating Iterator object may lead to another OutOfMemoryError, so + // we try to avoid any allocations + //noinspection ForLoopReplaceableByForEach + for (int i = 0; i < memoryResources.size(); i++) { + MemoryResource allocation = memoryResources.get(i); try { OS.memory().freeMemory(allocation.address, allocation.size); } catch (Throwable t) { - if (thrown == null) { - thrown = t; - } else { - thrown.addSuppressed(t); - } + thrown = Throwables.returnOrSuppress(thrown, t); } } return thrown; diff --git a/src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResources.java b/src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResources.java index 7a5bb99f2..7d14f084d 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResources.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/PersistedChronicleHashResources.java @@ -19,15 +19,13 @@ 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 net.openhft.chronicle.hash.impl.util.Throwables; import java.io.File; +import java.util.List; public final class PersistedChronicleHashResources extends ChronicleHashResources { - private static final Logger LOG = - LoggerFactory.getLogger(PersistedChronicleHashResources.class); private final File file; @@ -36,52 +34,42 @@ public final class PersistedChronicleHashResources extends ChronicleHashResource } @Override - public synchronized void run() { - if (memoryResources == null) - return; // Already released - // All Throwables are caught because this class is used as Runnable for sun.misc.Cleaner, - // hence must not fail + Throwable releaseSystemResources() { + Throwable thrown = null; + // Paranoiac mode: releaseMappedMemory() should never throw any Throwable (it should return + // it), but we don't trust ourselves and call releaseFile() in finally block. try { - Throwable thrown = doRelease(); - if (thrown != null) { - try { - LOG.error("Error on releasing memory or RAF of the 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(); - } - } + thrown = releaseMappedMemory(); } finally { - memoryResources = null; + thrown = releaseFile(thrown); } + return thrown; } - @Override - Throwable doRelease() { + private Throwable releaseMappedMemory() { Throwable thrown = null; - for (MemoryResource mapping : memoryResources) { + List memoryResources = memoryResources(); + // Indexed loop instead of for-each because we may be in the context of cleaning up after + // OutOfMemoryError, and allocating Iterator object may lead to another OutOfMemoryError, so + // we try to avoid any allocations + //noinspection ForLoopReplaceableByForEach + for (int i = 0; i < memoryResources.size(); i++) { + MemoryResource mapping = memoryResources.get(i); 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); + thrown = Throwables.returnOrSuppress(thrown, t); } } return thrown; } + + private Throwable releaseFile(Throwable thrown) { + try { + CanonicalRandomAccessFiles.release(file); + } catch (Throwable t) { + thrown = Throwables.returnOrSuppress(thrown, 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 cfeb67209..49ec33b72 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -24,9 +24,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.Throwables; 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; @@ -43,13 +41,16 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import sun.misc.Cleaner; +import java.io.Closeable; 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.*; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.concurrent.TimeUnit; import static java.lang.Long.numberOfTrailingZeros; @@ -161,13 +162,8 @@ public abstract class VanillaChronicleHash> allContexts; - /** * {@link ChronicleHashCloseOnExitHook} needs to use {@code VanillaChronicleHash}es as * WeakHashMap keys, but with identity comparison, not Map's equals() and hashCode(). @@ -391,8 +387,6 @@ public abstract class VanillaChronicleHash(); if (tierHashLookupSlotSize == 4) { @@ -403,7 +397,6 @@ public abstract class VanillaChronicleHash(); identity = new Identity(); } @@ -494,6 +487,10 @@ 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(this); - } - } - - resources.releaseManually(); + if (resources.releaseManually()) { // Releases nothing after resources.releaseManually(), only removes the cleaner // from the internal linked list of all cleaners. cleaner.clean(); ChronicleHashCloseOnExitHook.remove(this); - } catch (Throwable t) { - if (thrown == null) { - throw t; - } else { - thrown.addSuppressed(t); - throw Throwables.propagate(thrown); - } } } + protected void doClose(Throwable thrown) { + + } + @Override public boolean isOpen() { - return !closed; + return !resources.closed(); } public final void checkKey(Object key) { @@ -1047,41 +1012,16 @@ public abstract class VanillaChronicleHash(context)); - } + public void addCloseable(Closeable closeable) { + resources.addCloseable(closeable); } /** For testing only */ - public List> allContexts() { - return Collections.unmodifiableList(allContexts); + public List> allContexts() { + return Collections.unmodifiableList(resources.contexts()); } - - 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/hash/ThreadLocalState.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java index 0c674b7d6..a9f495861 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 @@ -81,14 +81,14 @@ public abstract class ThreadLocalState { MEMORY.writeOrderedInt(this, CONTEXT_LOCK_OFFSET, CONTEXT_UNLOCKED); } - public void closeContext(ChronicleHash hash) { + public void closeContext(String chronicleHashIdentityString) { if (tryCloseContext()) return; // If first attempt of closing a context (i. e. moving from unused to closed state) failed, // it means that the context is still in use. If this context belongs to the current thread, // this is a bug, because we cannot "wait" until context is unused in the same thread: if (owner() == Thread.currentThread()) { - throw new IllegalStateException(hash.toIdentityString() + + throw new IllegalStateException(chronicleHashIdentityString + ": Attempt to close a Chronicle Hash in the context " + "of not yet finished query or iteration"); } @@ -109,7 +109,7 @@ public abstract class ThreadLocalState { timeoutMillis--; } } while (timeoutMillis >= 0); - throw new RuntimeException(hash.toIdentityString() + + throw new RuntimeException(chronicleHashIdentityString + ": Failed to close a context, belonging to the thread\n" + owner() + ", in the state: " + owner().getState() + "\n" + "Possible reasons:\n" + diff --git a/src/main/java/net/openhft/chronicle/hash/impl/util/Throwables.java b/src/main/java/net/openhft/chronicle/hash/impl/util/Throwables.java index 733ccb44b..4bc75187a 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/util/Throwables.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/util/Throwables.java @@ -44,5 +44,15 @@ public final class Throwables { throw new RuntimeException(t); } + public static Throwable returnOrSuppress(Throwable thrown, Throwable t) { + if (thrown == null) { + return t; + } else { + if (t != null) + thrown.addSuppressed(t); + return thrown; + } + } + private Throwables() {} } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 2f158563b..ef1f798c5 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -1545,9 +1545,16 @@ public final class ChronicleMapBuilder implements return result; } catch (Throwable throwable) { try { - resources.releaseManually(); - } catch (Exception e) { - throwable.addSuppressed(e); + try { + resources.setChronicleHashIdentityString( + "ChronicleHash{name=" + name + ", file=" + file + "}"); + } catch (Throwable t) { + throwable.addSuppressed(t); + } finally { + resources.releaseManually(); + } + } catch (Throwable t) { + throwable.addSuppressed(t); } throw Throwables.propagateNotWrapping(throwable, IOException.class); } @@ -1555,6 +1562,7 @@ public final class ChronicleMapBuilder implements private void prepareMapPublication(VanillaChronicleMap map) throws IOException { establishReplication(map); + map.setResourcesName(); map.registerCleaner(); // Ensure safe publication of a ChronicleMap OS.memory().storeFence(); @@ -1729,9 +1737,16 @@ public final class ChronicleMapBuilder implements return map; } catch (Throwable throwable) { try { - resources.releaseManually(); - } catch (Exception e) { - throwable.addSuppressed(e); + try { + resources.setChronicleHashIdentityString( + "ChronicleHash{name=" + name + ", file=null}"); + } catch (Throwable t) { + throwable.addSuppressed(t); + } finally { + resources.releaseManually(); + } + } catch (Throwable t) { + throwable.addSuppressed(t); } throw Throwables.propagate(throwable); } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 978e6adcd..fb4f14fca 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -17,7 +17,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.algo.bitset.*; +import net.openhft.chronicle.algo.bitset.BitSetFrame; +import net.openhft.chronicle.algo.bitset.SingleThreadedFlatBitSetFrame; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.VanillaGlobalMutableState; @@ -34,11 +35,9 @@ import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.Closeable; import java.io.IOException; import java.util.Arrays; import java.util.Comparator; -import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReferenceArray; @@ -106,7 +105,6 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap closeables; /** * Idiomatically {@code assignedModificationIterators} should be a {@link CopyOnWriteArraySet}, @@ -181,7 +179,6 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap(128); - closeables = new CopyOnWriteArraySet<>(); tierModIterFrame = new SingleThreadedFlatBitSetFrame(computeTierModIterBitSetSizeInBits()); remoteNodeCouldBootstrapFrom = new long[128]; } @@ -259,26 +256,6 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap(this); - addContext(queryContext); - cxt.set(queryContext); - } - return queryContext; + + @Override + ChainingInterface newQueryContext() { + return new CompiledReplicatedMapQueryContext<>(this); } @Override @@ -599,15 +570,9 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap(this); - addContext(iterContext); - cxt.set(iterContext); - } - return iterContext; + @Override + ChainingInterface newIterationContext() { + return new CompiledReplicatedMapIterationContext<>(this); } public CompiledReplicatedMapIterationContext iterationContext() { diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index a6ccf43d5..f80cd15a1 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -22,6 +22,7 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.bytes.PointerBytesStore; import net.openhft.chronicle.core.io.Closeable; +import net.openhft.chronicle.hash.ChronicleHashClosedException; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.*; import net.openhft.chronicle.hash.impl.stage.entry.LocksInterface; @@ -68,6 +69,7 @@ public class VanillaChronicleMap ///////////////////////////////////////////////// private transient String name; + private transient String identityString; public transient boolean couldNotDetermineAlignmentBeforeAllocation; @@ -86,7 +88,7 @@ public class VanillaChronicleMap private transient boolean defaultEntryOperationsAndMethods; public transient DefaultValueProvider defaultValueProvider; - transient ThreadLocal cxt; + transient ThreadLocal cxt; public VanillaChronicleMap(ChronicleMapBuilder builder) throws IOException { super(builder); @@ -162,6 +164,7 @@ public class VanillaChronicleMap } private void initOwnTransients() { + identityString = makeIdentityString(); couldNotDetermineAlignmentBeforeAllocation = greatestCommonDivisor((int) chunkSize, alignment) != alignment; cxt = new ThreadLocal<>(); @@ -252,6 +255,10 @@ public class VanillaChronicleMap @Override public String toIdentityString() { + return identityString; + } + + private String makeIdentityString() { if (chronicleSet != null) return chronicleSet.toIdentityString(); return "ChronicleMap{" + @@ -283,15 +290,25 @@ public class VanillaChronicleMap return (addr + alignment - 1) & ~(alignment - 1L); } - private ChainingInterface q() { - //noinspection unchecked - ChainingInterface queryContext = cxt.get(); - if (queryContext == null) { - queryContext = new CompiledMapQueryContext<>(this); - addContext(queryContext); - cxt.set(queryContext); + final ChainingInterface q() { + ContextHolder contextHolder = cxt.get(); + if (contextHolder == null) { + ChainingInterface queryContext = newQueryContext(); + contextHolder = new ContextHolder(queryContext); + addContext(contextHolder); + cxt.set(contextHolder); + return queryContext; } - return queryContext; + ChainingInterface queryContext = contextHolder.get(); + if (queryContext != null) { + return queryContext; + } else { + throw new ChronicleHashClosedException(this); + } + } + + ChainingInterface newQueryContext() { + return new CompiledMapQueryContext<>(this); } public QueryContextInterface mapContext() { @@ -303,15 +320,25 @@ public class VanillaChronicleMap (c, m) -> new CompiledMapQueryContext(c, m), this); } - private ChainingInterface i() { - //noinspection unchecked - ChainingInterface iterContext = cxt.get(); - if (iterContext == null) { - iterContext = new CompiledMapIterationContext<>(this); - addContext(iterContext); - cxt.set(iterContext); + final ChainingInterface i() { + ContextHolder contextHolder = cxt.get(); + if (contextHolder == null) { + ChainingInterface iterationContext = newIterationContext(); + contextHolder = new ContextHolder(iterationContext); + addContext(contextHolder); + cxt.set(contextHolder); + return iterationContext; } - return iterContext; + ChainingInterface iterContext = contextHolder.get(); + if (iterContext != null) { + return iterContext; + } else { + throw new ChronicleHashClosedException(this); + } + } + + ChainingInterface newIterationContext() { + return new CompiledMapIterationContext<>(this); } public IterationContext iterationContext() { diff --git a/src/test/java/net/openhft/chronicle/map/MapCloseTest.java b/src/test/java/net/openhft/chronicle/map/MapCloseTest.java index 6169fce14..e3b56381a 100644 --- a/src/test/java/net/openhft/chronicle/map/MapCloseTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapCloseTest.java @@ -120,7 +120,7 @@ public class MapCloseTest { map.get(1); Assert.assertEquals(1, map.allContexts().size()); - ChainingInterface cxt = map.allContexts().get(0).get(); + ChainingInterface cxt = map.allContexts().get(0).get().get(); Assert.assertTrue(cxt == map.queryContext(1)); } }