From d04b7cfc207436860787315902e8f1d6872fcc9a Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 1 Oct 2016 20:00:49 +0300 Subject: [PATCH] Make OldDeletedEntriesCleanup a Thread instead of Runnable to be executed in a single-threaded executor --- .../chronicle/map/ChronicleMapBuilder.java | 28 ++++++----------- ...va => OldDeletedEntriesCleanupThread.java} | 31 ++++++++++++------- 2 files changed, 29 insertions(+), 30 deletions(-) rename src/main/java/net/openhft/chronicle/map/{OldDeletedEntriesCleanup.java => OldDeletedEntriesCleanupThread.java} (86%) diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 861a3de48..a0a4ac478 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -36,7 +36,6 @@ import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.hash.serialization.impl.SerializationBuilder; import net.openhft.chronicle.map.replication.MapRemoteOperations; import net.openhft.chronicle.set.ChronicleSetBuilder; -import net.openhft.chronicle.threads.NamedThreadFactory; import net.openhft.chronicle.values.ValueModel; import net.openhft.chronicle.values.Values; import net.openhft.chronicle.wire.TextWire; @@ -56,8 +55,6 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import static java.lang.Double.isNaN; @@ -124,10 +121,10 @@ public final class ChronicleMapBuilder implements private static final int NO_ALIGNMENT = 1; /** - * If want to increase this number, note {@link OldDeletedEntriesCleanup} uses array to store - * all segment indexes -- so it could be current JVM max array size, not Integer.MAX_VALUE - * (which is an obvious limitation, as many APIs and internals use int type for representing - * segment index). + * If want to increase this number, note {@link OldDeletedEntriesCleanupThread} uses array + * to store all segment indexes -- so it could be current JVM max array size, + * not Integer.MAX_VALUE (which is an obvious limitation, as many APIs and internals use int + * type for representing segment index). * * Anyway, unlikely anyone ever need more than 1 billion segments. */ @@ -1729,20 +1726,15 @@ public final class ChronicleMapBuilder implements } private void establishCleanupThread(ReplicatedChronicleMap map) { - OldDeletedEntriesCleanup cleanup = new OldDeletedEntriesCleanup(map); - NamedThreadFactory threadFactory = - new NamedThreadFactory("cleanup thread for map persisted at " + map.file()); - ExecutorService executor = Executors.newSingleThreadExecutor(threadFactory); - executor.submit(cleanup); - - map.addCloseable(cleanup); + OldDeletedEntriesCleanupThread cleanupThread = new OldDeletedEntriesCleanupThread(map); + map.addCloseable(cleanupThread); + cleanupThread.start(); // WARNING this relies on the fact that ReplicatedChronicleMap closes closeables in the same - // order as they are added, i. e. OldDeletedEntriesCleanup instance close()d before the - // following closeable + // order as they are added, i. e. OldDeletedEntriesCleanupThread instance close()d before + // the following closeable map.addCloseable(() -> { - executor.shutdown(); try { - executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + cleanupThread.join(); } catch (InterruptedException e) { LOG.error("", e); } diff --git a/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupThread.java similarity index 86% rename from src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java rename to src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupThread.java index 1bc0b3007..6afbeb218 100644 --- a/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java +++ b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanupThread.java @@ -32,18 +32,28 @@ import java.util.function.Predicate; import static net.openhft.chronicle.hash.replication.TimeProvider.currentTime; import static net.openhft.chronicle.hash.replication.TimeProvider.systemTimeIntervalBetween; -class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate { - private static final Logger LOG = LoggerFactory.getLogger(OldDeletedEntriesCleanup.class); +class OldDeletedEntriesCleanupThread extends Thread + implements Closeable, Predicate { + private static final Logger LOG = LoggerFactory.getLogger(OldDeletedEntriesCleanupThread.class); private final ReplicatedChronicleMap map; private final int[] segmentsPermutation; private final int[] inverseSegmentsPermutation; + + /** + * This object is used to determine that this thread is parked from {@link #sleepMillis(long)} + * or {@link #sleepNanos(long)}, not somewhere inside ChronicleMap logic, to interrupt() + * selectively in {@link #close()}. + */ + private final Object cleanupSleepingHandle = new Object(); + private volatile boolean shutdown; - private volatile Thread runnerThread; + private long prevSegment0ScanStart = -1; private long removedCompletely; - public OldDeletedEntriesCleanup(ReplicatedChronicleMap map) { + public OldDeletedEntriesCleanupThread(ReplicatedChronicleMap map) { + super("Cleanup Thread for ReplicatedChronicleMap persisted to " + map.file()); this.map = map; segmentsPermutation = randomPermutation(map.segments()); inverseSegmentsPermutation = inversePermutation(segmentsPermutation); @@ -51,7 +61,6 @@ class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate