From e0b534af23e2cbc3e8210252dd15cfc038950cfd Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 10 Dec 2016 20:00:41 -0600 Subject: [PATCH] Move all ChronicleMap closing logic into independent ChronicleHashResources. Reference contexts from ThreadLocals via ContextHolders and clear references to contexts on closing ChronicleMap. Make some closing procedures more paranoid than they were. Cache ChronicleMap's identityString and always return the same String from toIdentityString(). Add Throwables.returnOrSuppress(). --- .../hash/ChronicleHashClosedException.java | 6 +- .../hash/impl/ChronicleHashResources.java | 219 ++++++++++++++++-- .../chronicle/hash/impl/ContextHolder.java | 69 ++++++ .../impl/InMemoryChronicleHashResources.java | 47 +--- .../impl/PersistedChronicleHashResources.java | 66 +++--- .../hash/impl/VanillaChronicleHash.java | 100 ++------ .../impl/stage/hash/ThreadLocalState.java | 6 +- .../chronicle/hash/impl/util/Throwables.java | 10 + .../chronicle/map/ChronicleMapBuilder.java | 27 ++- .../chronicle/map/ReplicatedChronicleMap.java | 53 +---- .../chronicle/map/VanillaChronicleMap.java | 61 +++-- .../openhft/chronicle/map/MapCloseTest.java | 2 +- 12 files changed, 424 insertions(+), 242 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/ContextHolder.java 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)); } }