Standardize how ReferenceCounted resources are managed and traced, https://github.com/OpenHFT/Chronicle-Core/issues/121

pom.xml

@@ -10,7 +10,7 @@
 
     <modelVersion>4.0.0</modelVersion>
     <artifactId>chronicle-map</artifactId>
-    <version>3.19.32-SNAPSHOT</version>
+    <version>3.19.40-SNAPSHOT</version>
     <name>OpenHFT/Chronicle-Map</name>
     <description>Chronicle-Map</description>
     <packaging>bundle</packaging>

src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java

@@ -524,11 +524,11 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
         key.writeTo(destination, destination.writePosition());
         destination.writeSkip(key.size());
 
-        boolean debugEnabled = LOG.isDebugEnabled();
+        boolean traceEnabled = LOG.isTraceEnabled();
         String message = null;
-        if (debugEnabled) {
+        if (traceEnabled) {
             if (isDeleted) {
-                LOG.debug("WRITING ENTRY TO DEST -  into local-id={}, remove(key={})",
+                LOG.trace("WRITING ENTRY TO DEST -  into local-id={}, remove(key={})",
                         identifier(), key);
             } else {
                 message = String.format(
@@ -545,7 +545,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
         value.writeTo(destination, destination.writePosition());
         destination.writeSkip(value.size());
 
-        if (debugEnabled) {
+        if (traceEnabled) {
             LOG.debug(message + "value=" + value + ")");
         }
     }

src/main/java/net/openhft/chronicle/map/impl/stage/data/ZeroBytesStore.java

@@ -19,6 +19,7 @@ package net.openhft.chronicle.map.impl.stage.data;
 import net.openhft.chronicle.bytes.BytesStore;
 import net.openhft.chronicle.bytes.RandomDataInput;
 import net.openhft.chronicle.core.OS;
+import net.openhft.chronicle.core.io.ReferenceOwner;
 import org.jetbrains.annotations.Nullable;
 
 import java.nio.BufferOverflowException;
@@ -37,6 +38,7 @@ public enum ZeroBytesStore implements BytesStore<ZeroBytesStore, Void> {
         throw new UnsupportedOperationException("Cannot get the address of nothing");
     }
 
+
     @Override
     public int peekUnsignedByte(long offset) {
         return 0;
@@ -155,24 +157,34 @@ public enum ZeroBytesStore implements BytesStore<ZeroBytesStore, Void> {
         return true;
     }
 
+
     @Override
-    public void reserve() throws IllegalStateException {
+    public void reserve(ReferenceOwner id) throws IllegalStateException {
     }
 
     @Override
-    public void release() throws IllegalStateException {
+    public boolean tryReserve(ReferenceOwner id) throws IllegalStateException {
+        return false;
     }
 
     @Override
-    public long refCount() {
+    public boolean reservedBy(ReferenceOwner owner) {
+        return true;
+    }
+
+    @Override
+    public void release(ReferenceOwner id) throws IllegalStateException {
+    }
+
+    @Override
+    public void releaseLast(ReferenceOwner id) throws IllegalStateException {
+    }
+
+    @Override
+    public int refCount() {
         return 0;
     }
 
-    @Override
-    public boolean tryReserve() {
-        throw new UnsupportedOperationException("todo");
-    }
-
     @Override
     public ZeroBytesStore writeByte(long l, byte b) {
         throw new UnsupportedOperationException();

src/test/java/examples/portfolio/PortfolioValueTest.java

@@ -20,6 +20,7 @@ import net.openhft.chronicle.core.values.LongValue;
 import net.openhft.chronicle.map.ChronicleMap;
 import net.openhft.chronicle.map.ChronicleMapBuilder;
 import net.openhft.chronicle.map.MapSegmentContext;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import net.openhft.chronicle.values.Values;
 import org.apache.commons.lang3.mutable.MutableDouble;
 import org.junit.Test;
@@ -42,7 +43,8 @@ public class PortfolioValueTest {
     private static void computeValueUsingIterator(final ChronicleMap<LongValue, PortfolioAssetInterface> cache) throws ExecutionException, InterruptedException {
         long startTime = System.currentTimeMillis();
 
-        ExecutorService executor = Executors.newFixedThreadPool(nThreads);
+        ExecutorService executor = Executors.newFixedThreadPool(nThreads,
+                new NamedThreadFactory("test"));
         @SuppressWarnings("unchecked")
         Future<Double>[] futures = new Future[nThreads];
         long batchSize = (useIterator ? cache.segments() : nAssets) / nThreads;
@@ -118,7 +120,8 @@ public class PortfolioValueTest {
     private void createData(final ChronicleMap<LongValue, PortfolioAssetInterface> cache) throws ExecutionException, InterruptedException {
         long startTime = System.currentTimeMillis();
 
-        ExecutorService executor = Executors.newFixedThreadPool(nThreads);
+        ExecutorService executor = Executors.newFixedThreadPool(nThreads,
+                new NamedThreadFactory("test"));
         Future<?>[] futures = new Future[nThreads];
         long batchSize = nAssets / nThreads;
 

src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java

@@ -18,6 +18,7 @@ package net.openhft.chronicle.map;
 
 import net.openhft.chronicle.bytes.Bytes;
 import net.openhft.chronicle.hash.replication.ReplicableEntry;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import org.hamcrest.CoreMatchers;
 import org.junit.Test;
 
@@ -65,7 +66,8 @@ public class BasicReplicationTest {
             processorThree.addDestinationMap(mapThree.acquireModificationIterator(asByte(1)), mapThree, mapOne);
             processorThree.addDestinationMap(mapThree.acquireModificationIterator(asByte(2)), mapThree, mapTwo);
 
-            final ExecutorService executorService = Executors.newFixedThreadPool(3);
+            final ExecutorService executorService = Executors.newFixedThreadPool(3,
+                    new NamedThreadFactory("test"));
             executorService.submit(processorOne::processPendingChangesLoop);
             executorService.submit(processorTwo::processPendingChangesLoop);
             executorService.submit(processorThree::processPendingChangesLoop);
@@ -175,7 +177,8 @@ public class BasicReplicationTest {
         private final ReplicatedChronicleMap<K, V, ?> sourceMap;
         private final ReplicatedChronicleMap<K, V, ?> destinationMap;
         private final Bytes<ByteBuffer> buffer = Bytes.elasticByteBuffer(4096);
-        private final ExecutorService delayedExecutor = Executors.newSingleThreadExecutor();
+        private final ExecutorService delayedExecutor = Executors.newSingleThreadExecutor(
+                new NamedThreadFactory("delayed"));
         private final AtomicInteger messagesInflight = new AtomicInteger(0);
         private final ArrayList<String> keys = new ArrayList<>();
 

src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java

@@ -17,6 +17,7 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.chronicle.core.Jvm;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import org.junit.Test;
 
 import java.io.File;
@@ -47,10 +48,12 @@ public class ChronicleMapSanityCheckTest {
                 file.getAbsolutePath().toString());
 
         ScheduledExecutorService producerExecutor =
-                Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors() - 1);
+                Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors() - 1,
+                        new NamedThreadFactory("producer"));
 
         ScheduledExecutorService consumerExecutor =
-                Executors.newSingleThreadScheduledExecutor();
+                Executors.newSingleThreadScheduledExecutor(
+                        new NamedThreadFactory("consumer"));
 
         int N = 1000;
 

src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java

@@ -23,6 +23,7 @@ import net.openhft.chronicle.bytes.BytesStore;
 import net.openhft.chronicle.core.values.IntValue;
 import net.openhft.chronicle.core.values.LongValue;
 import net.openhft.chronicle.set.Builder;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import net.openhft.chronicle.values.Values;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -917,7 +918,7 @@ public class ChronicleMapTest {
 //        int runs = Integer.getInteger("runs", 10);
         int procs = 1; // Runtime.getRuntime().availableProcessors();
         int threads = procs * 3;
-        ExecutorService es = Executors.newFixedThreadPool(procs);
+        ExecutorService es = Executors.newFixedThreadPool(procs, new NamedThreadFactory("test"));
         for (int runs : new int[]{1, /*10, 250, 500, 1000, 2500*/}) {
             for (int entrySize : new int[]{240, 256}) {
                 int valuePadding = entrySize - 16;
@@ -1009,7 +1010,7 @@ public class ChronicleMapTest {
         int procs = Runtime.getRuntime().availableProcessors();
         if (procs > 8) procs--;
         int threads = procs * 3;
-        ExecutorService es = Executors.newFixedThreadPool(procs);
+        ExecutorService es = Executors.newFixedThreadPool(procs, new NamedThreadFactory("test"));
         for (int runs : new int[]{1, 2, 5, 10, 25, 50, 100, 500, 1000, 2500}) {
 
             final long entries = runs * 1000 * 1000L;
@@ -1106,7 +1107,7 @@ public class ChronicleMapTest {
 //        int runs = Integer.getInteger("runs", 10);
         int procs = Runtime.getRuntime().availableProcessors();
         int threads = procs * 3; // runs > 100 ? procs / 2 : procs;
-        ExecutorService es = Executors.newFixedThreadPool(procs);
+        ExecutorService es = Executors.newFixedThreadPool(procs, new NamedThreadFactory("test"));
         for (int runs : new int[]{10, 50, 100, 250, 500, 1000, 2500}) {
 // JAVA 8 produces more garbage than previous versions for internal work.
 //            System.gc();
@@ -1185,7 +1186,7 @@ public class ChronicleMapTest {
             for (int j = 0; j < count; j++) {
                 final ConcurrentMap<String, AtomicInteger> map = new ConcurrentHashMap<String, AtomicInteger>((int) (entries * 5 / 4), 1.0f, 1024);
                 long start = System.currentTimeMillis();
-                ExecutorService es = Executors.newFixedThreadPool(procs);
+                ExecutorService es = Executors.newFixedThreadPool(procs, new NamedThreadFactory("test"));
                 for (int i = 0; i < threads; i++) {
                     final int t = i;
                     es.submit(new Runnable() {

src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java

@@ -17,6 +17,7 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.chronicle.core.values.LongValue;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -116,7 +117,7 @@ public class EntryCountMapTest {
     public void testSmall() throws IOException, ExecutionException, InterruptedException {
         System.out.print("testSmall seeds");
         int procs = Runtime.getRuntime().availableProcessors();
-        ExecutorService es = Executors.newFixedThreadPool(procs);
+        ExecutorService es = Executors.newFixedThreadPool(procs, new NamedThreadFactory("test"));
         for (int t = 0; t < ecmTests; t++) {
             List<Future<?>> futures = new ArrayList<>();
             {
@@ -160,7 +161,7 @@ public class EntryCountMapTest {
     public void testMedium() throws IOException, ExecutionException, InterruptedException {
         System.out.print("testMedium seeds");
         int procs = Runtime.getRuntime().availableProcessors();
-        ExecutorService es = Executors.newFixedThreadPool(procs);
+        ExecutorService es = Executors.newFixedThreadPool(procs, new NamedThreadFactory("test"));
         for (int t = 0; t < ecmTests; t++) {
             // regression test.
             List<Future<?>> futures = new ArrayList<>();

src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java

@@ -16,6 +16,7 @@
 
 package net.openhft.chronicle.map;
 
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -51,7 +52,7 @@ public class LargeEntriesTest {
                 .createPersistedTo(file)) {
             warmUpCompression(ENTRY_SIZE);
             int threads = 4; //Runtime.getRuntime().availableProcessors();
-            ExecutorService es = Executors.newFixedThreadPool(threads);
+            ExecutorService es = Executors.newFixedThreadPool(threads, new NamedThreadFactory("test"));
             final int block = ENTRIES / threads;
             for (int i = 0; i < 3; i++) {
                 long start = System.currentTimeMillis();
@@ -119,7 +120,8 @@ public class LargeEntriesTest {
         {
 //            warmUpCompression(ENTRY_SIZE);
             int threads = Runtime.getRuntime().availableProcessors();
-            ExecutorService es = Executors.newFixedThreadPool(threads);
+            ExecutorService es = Executors.newFixedThreadPool(threads,
+                    new NamedThreadFactory("test"));
             final int block = ENTRIES / threads;
             for (int i = 0; i < 3; i++) {
                 long start = System.currentTimeMillis();

src/test/java/net/openhft/chronicle/map/RecoverTest.java

@@ -20,6 +20,7 @@ import net.openhft.chronicle.core.values.LongValue;
 import net.openhft.chronicle.hash.ChecksumEntry;
 import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
 import net.openhft.chronicle.hash.ChronicleHashCorruption;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import net.openhft.chronicle.values.Values;
 import org.junit.Assert;
 import org.junit.Ignore;
@@ -69,7 +70,8 @@ public class RecoverTest {
         // acquires read lock successfully
         assertNull(map.get(0));
 
-        ExecutorService executorService = Executors.newSingleThreadExecutor();
+        ExecutorService executorService = Executors.newSingleThreadExecutor(
+                new NamedThreadFactory("recoverTest"));
 
         executorService.submit(() -> {
             ExternalMapQueryContext<Integer, Integer, ?> c = map.queryContext(0);

src/test/java/net/openhft/chronicle/map/TrickyContextCasesTest.java

@@ -17,6 +17,7 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.chronicle.core.values.IntValue;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import net.openhft.chronicle.values.Values;
 import org.junit.Test;
 
@@ -65,14 +66,16 @@ public class TrickyContextCasesTest {
         try (ExternalMapQueryContext<Integer, byte[], ?> q = map.queryContext(1)) {
             MapEntry<Integer, byte[]> entry = q.entry(); // acquires read lock implicitly
             assertNotNull(entry);
-            Executors.newFixedThreadPool(1).submit(() -> {
-                // this call should try to acquire write lock, that should lead to dead lock
-                // but if not...
-                // relocates the entry for the key 1 after the entry for 2, under update lock
-                map.put(1, new byte[]{1, 2, 3, 4, 5});
-                // puts the entry for 3 at the place of the entry for the key 1, under update lock
-                map.put(3, new byte[]{3});
-            }).get();
+            Executors.newFixedThreadPool(1,
+                    new NamedThreadFactory("test"))
+                    .submit(() -> {
+                        // this call should try to acquire write lock, that should lead to dead lock
+                        // but if not...
+                        // relocates the entry for the key 1 after the entry for 2, under update lock
+                        map.put(1, new byte[]{1, 2, 3, 4, 5});
+                        // puts the entry for 3 at the place of the entry for the key 1, under update lock
+                        map.put(3, new byte[]{3});
+                    }).get();
             // prints [3]
             System.out.println(Arrays.toString(entry.value().get()));
         }

src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java

@@ -18,6 +18,7 @@ package net.openhft.chronicle.map.example;
 
 import net.openhft.chronicle.map.ChronicleMap;
 import net.openhft.chronicle.map.ChronicleMapBuilder;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import net.openhft.chronicle.values.Array;
 import net.openhft.chronicle.values.Values;
 
@@ -59,7 +60,8 @@ public class LotsOfEntriesMain {
                 .averageKeySize((Math.log(1.024) - Math.log(0.024)) * 24 + 2)
                 .createPersistedTo(file);
         int threads = Runtime.getRuntime().availableProcessors();
-        ExecutorService es = Executors.newFixedThreadPool(threads);
+        ExecutorService es = Executors.newFixedThreadPool(threads,
+                new NamedThreadFactory("test"));
         long block = (entries + threads - 1) / threads;
 
         final long start = System.nanoTime();

src/test/java/net/openhft/chronicle/map/perf/MapStress.java

@@ -1,6 +1,7 @@
 package net.openhft.chronicle.map.perf;
 
 import net.openhft.chronicle.map.ChronicleMap;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 
 import java.io.File;
 import java.io.IOException;
@@ -20,7 +21,8 @@ public class MapStress {
     private ChronicleMap<String, Security> map = ChronicleMap.of(String.class, Security.class).averageKeySize(500).averageValueSize(500).entries(100_000).createOrRecoverPersistedTo(file);
     private ChronicleMap<String, Security> readerMap = ChronicleMap.of(String.class, Security.class).averageKeySize(50).averageValueSize(500).entries(100_000).createOrRecoverPersistedTo(file);
 
-    private ExecutorService executor = Executors.newFixedThreadPool(N_WRITE_THREADS + N_READ_THREADS);
+    private ExecutorService executor = Executors.newFixedThreadPool(N_WRITE_THREADS + N_READ_THREADS,
+            new NamedThreadFactory("mapStress"));
     private ThreadLocal<Random> randomTL = ThreadLocal.withInitial(Random::new);
 
     public MapStress() throws IOException {

src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java

@@ -18,6 +18,7 @@ package net.openhft.chronicle.set;
 
 import net.openhft.chronicle.hash.ChronicleHash;
 import net.openhft.chronicle.hash.ChronicleHashBuilder;
+import net.openhft.chronicle.threads.NamedThreadFactory;
 import org.junit.Test;
 
 import java.io.File;
@@ -55,7 +56,8 @@ public class Issue24ChronicleSetTest {
     @Test
     public void issue24ChronicleSetTest() throws IOException {
         ChronicleSet<String> set = initSet(String.class, 1_000_000, 30);
-        ExecutorService executor = Executors.newFixedThreadPool(5);
+        ExecutorService executor = Executors.newFixedThreadPool(5,
+                new NamedThreadFactory("test"));
         for (int i = 0; i < 10; i++) {
             Runnable worker = new WorkerThread(set);
             executor.execute(worker);

system.properties

@@ -0,0 +1,4 @@
+# Tracing if resources are closed/released correctly.
+jvm.resource.tracing=true
+# for profiling
+jvm.safepoint.enabled=false