From 12e286f9ec539eb8ca397d745c685a3bd9b68a12 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Thu, 11 Jun 2020 19:51:48 +0100 Subject: [PATCH] Standardize how ReferenceCounted resources are managed and traced, https://github.com/OpenHFT/Chronicle-Core/issues/121 --- pom.xml | 2 +- .../chronicle/map/ReplicatedChronicleMap.java | 8 +++--- .../map/impl/stage/data/ZeroBytesStore.java | 28 +++++++++++++------ .../portfolio/PortfolioValueTest.java | 7 +++-- .../chronicle/map/BasicReplicationTest.java | 7 +++-- .../map/ChronicleMapSanityCheckTest.java | 7 +++-- .../chronicle/map/ChronicleMapTest.java | 9 +++--- .../chronicle/map/EntryCountMapTest.java | 5 ++-- .../chronicle/map/LargeEntriesTest.java | 6 ++-- .../openhft/chronicle/map/RecoverTest.java | 4 ++- .../chronicle/map/TrickyContextCasesTest.java | 19 +++++++------ .../map/example/LotsOfEntriesMain.java | 4 ++- .../openhft/chronicle/map/perf/MapStress.java | 4 ++- .../set/Issue24ChronicleSetTest.java | 4 ++- system.properties | 4 +++ 15 files changed, 79 insertions(+), 39 deletions(-) create mode 100644 system.properties diff --git a/pom.xml b/pom.xml index fb97e2b06..046fbedb0 100755 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ 4.0.0 chronicle-map - 3.19.32-SNAPSHOT + 3.19.40-SNAPSHOT OpenHFT/Chronicle-Map Chronicle-Map bundle diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index c1cec0737..a551cdc16 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -524,11 +524,11 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap { 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 { 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(); diff --git a/src/test/java/examples/portfolio/PortfolioValueTest.java b/src/test/java/examples/portfolio/PortfolioValueTest.java index 3beba449e..64b8c59bd 100644 --- a/src/test/java/examples/portfolio/PortfolioValueTest.java +++ b/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 cache) throws ExecutionException, InterruptedException { long startTime = System.currentTimeMillis(); - ExecutorService executor = Executors.newFixedThreadPool(nThreads); + ExecutorService executor = Executors.newFixedThreadPool(nThreads, + new NamedThreadFactory("test")); @SuppressWarnings("unchecked") Future[] futures = new Future[nThreads]; long batchSize = (useIterator ? cache.segments() : nAssets) / nThreads; @@ -118,7 +120,8 @@ public class PortfolioValueTest { private void createData(final ChronicleMap 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; diff --git a/src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java b/src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java index 9dc28723b..a22a4bc0b 100644 --- a/src/test/java/net/openhft/chronicle/map/BasicReplicationTest.java +++ b/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 sourceMap; private final ReplicatedChronicleMap destinationMap; private final Bytes 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 keys = new ArrayList<>(); diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java index 28c02ef66..8a12f33f1 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java +++ b/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; diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index f6f3d4034..792964e22 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/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 map = new ConcurrentHashMap((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() { diff --git a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java b/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java index 698dc5246..9f4c4236b 100644 --- a/src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java +++ b/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> 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> futures = new ArrayList<>(); diff --git a/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java b/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java index c3fb73ee4..fc5b9fe61 100755 --- a/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java +++ b/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(); diff --git a/src/test/java/net/openhft/chronicle/map/RecoverTest.java b/src/test/java/net/openhft/chronicle/map/RecoverTest.java index abf39dece..f6c986d41 100644 --- a/src/test/java/net/openhft/chronicle/map/RecoverTest.java +++ b/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 c = map.queryContext(0); diff --git a/src/test/java/net/openhft/chronicle/map/TrickyContextCasesTest.java b/src/test/java/net/openhft/chronicle/map/TrickyContextCasesTest.java index f194c6a5d..82210538d 100644 --- a/src/test/java/net/openhft/chronicle/map/TrickyContextCasesTest.java +++ b/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 q = map.queryContext(1)) { MapEntry 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())); } diff --git a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java b/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java index 4d57453d1..a33bb60ca 100755 --- a/src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java +++ b/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(); diff --git a/src/test/java/net/openhft/chronicle/map/perf/MapStress.java b/src/test/java/net/openhft/chronicle/map/perf/MapStress.java index dc67677a9..12543389d 100644 --- a/src/test/java/net/openhft/chronicle/map/perf/MapStress.java +++ b/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 map = ChronicleMap.of(String.class, Security.class).averageKeySize(500).averageValueSize(500).entries(100_000).createOrRecoverPersistedTo(file); private ChronicleMap 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 randomTL = ThreadLocal.withInitial(Random::new); public MapStress() throws IOException { diff --git a/src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java b/src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java index eb8a1b2e3..48b76c5f6 100644 --- a/src/test/java/net/openhft/chronicle/set/Issue24ChronicleSetTest.java +++ b/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 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); diff --git a/system.properties b/system.properties new file mode 100644 index 000000000..677f6b473 --- /dev/null +++ b/system.properties @@ -0,0 +1,4 @@ +# Tracing if resources are closed/released correctly. +jvm.resource.tracing=true +# for profiling +jvm.safepoint.enabled=false \ No newline at end of file