Make OldDeletedEntriesCleanup a Thread instead of Runnable to be executed in a single-threaded executor

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<K, V> 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<K, V> 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);
             }

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<ReplicableEntry> {
-    private static final Logger LOG = LoggerFactory.getLogger(OldDeletedEntriesCleanup.class);
+class OldDeletedEntriesCleanupThread extends Thread
+        implements Closeable, Predicate<ReplicableEntry> {
+    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<Replica
 
     @Override
     public void run() {
-        runnerThread = Thread.currentThread();
         while (!shutdown) {
             int segmentIndex = map.globalMutableState().getCurrentCleanupSegmentIndex();
             int nextSegmentIndex;
@@ -108,23 +117,21 @@ class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate<Replica
     private void sleepMillis(long millis) {
         long deadline = System.currentTimeMillis() + millis;
         while (System.currentTimeMillis() < deadline && !shutdown)
-            LockSupport.parkUntil(this, deadline);
+            LockSupport.parkUntil(cleanupSleepingHandle, deadline);
     }
 
     private void sleepNanos(long nanos) {
         long deadline = System.nanoTime() + nanos;
         while (System.nanoTime() < deadline && !shutdown)
-            LockSupport.parkNanos(this, deadline);
+            LockSupport.parkNanos(cleanupSleepingHandle, deadline);
     }
 
     @Override
     public void close() {
         shutdown = true;
-        if (runnerThread != null &&
-                // this means blocked in sleepMillis() or sleepNanos()
-                LockSupport.getBlocker(runnerThread) == this) {
-            runnerThread.interrupt(); // unblock
-        }
+        // this means blocked in sleepMillis() or sleepNanos()
+        if (LockSupport.getBlocker(this) == cleanupSleepingHandle)
+            this.interrupt(); // unblock
     }
 
     private int nextSegmentIndex(int segmentIndex) {