Stop threads and subscribers that outlive a plugin disable (#1020)

* Stop threads and subscribers that outlive a plugin disable

The WorldEdit subscriber, the cache sweeper thread, per-click inspector threads and the bStats scheduler all kept running after CoreProtect was disabled, and the Folia shutdown checkpoint could wait 30 seconds on tasks that never run. Shutdown now unregisters and stops each of them, inspector lookups use a small named pool, and the checkpoint waits 5 seconds.

* Recreate inspector executor on enable and account for shutdown leftovers

The inspector executor was a static final that stayed shut down after a disable, so a same-instance enable could never run another lookup. It is now created on each enable. At shutdown, lookups still queued after the drain window release their throttle slot, since shutdownNow() returns them without running their finally block, and running lookups that survive the interrupt are logged by thread and player after a bounded wait. The Folia checkpoint timeout change is reverted to keep this change to thread cleanup.

src/main/java/net/coreprotect/listener/entity/HangingBreakByEntityListener.java

@@ -26,6 +26,7 @@ import net.coreprotect.database.Database;
 import net.coreprotect.database.lookup.BlockLookup;
 import net.coreprotect.language.Phrase;
 import net.coreprotect.listener.player.PlayerInteractEntityListener;
+import net.coreprotect.listener.player.inspector.BaseInspector;
 import net.coreprotect.utility.Chat;
 import net.coreprotect.utility.Color;
 import net.coreprotect.utility.MaterialUtils;
@@ -90,15 +91,7 @@ public final class HangingBreakByEntityListener extends Queue implements Listene
                 }
             }
         }
-        try {
-            Runnable runnable = new BasicThread();
-            Thread thread = new Thread(runnable);
-            thread.start();
-        }
-        catch (RuntimeException | Error e) {
-            LookupThrottle.release(player.getName());
-            throw e;
-        }
+        BaseInspector.runLookup(player, new BasicThread());
     }
 
     @EventHandler(priority = EventPriority.HIGHEST)

src/main/java/net/coreprotect/listener/player/ArmorStandManipulateListener.java

@@ -21,6 +21,7 @@ import net.coreprotect.consumer.Queue;
 import net.coreprotect.database.Database;
 import net.coreprotect.database.lookup.ChestTransactionLookup;
 import net.coreprotect.language.Phrase;
+import net.coreprotect.listener.player.inspector.BaseInspector;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.utility.Chat;
 import net.coreprotect.utility.Color;
@@ -73,15 +74,7 @@ public final class ArmorStandManipulateListener extends Queue implements Listene
                 }
             }
         }
-        try {
-            Runnable runnable = new BasicThread();
-            Thread thread = new Thread(runnable);
-            thread.start();
-        }
-        catch (RuntimeException | Error e) {
-            LookupThrottle.release(finalPlayer.getName());
-            throw e;
-        }
+        BaseInspector.runLookup(finalPlayer, new BasicThread());
     }
 
     @EventHandler(priority = EventPriority.MONITOR)

src/main/java/net/coreprotect/listener/player/inspector/BaseInspector.java

@@ -1,6 +1,14 @@
 package net.coreprotect.listener.player.inspector;
 
 import java.sql.Connection;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.bukkit.entity.Player;
 
@@ -13,6 +21,29 @@ import net.coreprotect.utility.LookupThrottle;
 
 public abstract class BaseInspector {
 
+    private static final int LOOKUP_THREADS = 2;
+    private static final int LOOKUP_QUEUE_SIZE = 64;
+    private static final long LOOKUP_THREAD_IDLE_SECONDS = 30L;
+    private static final long SHUTDOWN_WAIT_SECONDS = 5L;
+    private static final long INTERRUPT_WAIT_SECONDS = 2L;
+    private static final AtomicInteger lookupThreadCount = new AtomicInteger();
+    private static final Set<InspectorLookup> activeLookups = ConcurrentHashMap.newKeySet();
+    private static volatile ThreadPoolExecutor lookupExecutor;
+
+    /**
+     * Creates the inspector threads for this enable. Each enable gets a new executor because a disable shuts the
+     * previous one down.
+     */
+    public static void startLookups() {
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(LOOKUP_THREADS, LOOKUP_THREADS, LOOKUP_THREAD_IDLE_SECONDS, TimeUnit.SECONDS, new ArrayBlockingQueue<>(LOOKUP_QUEUE_SIZE), runnable -> {
+            Thread thread = new Thread(runnable, "CoreProtect-Inspector-" + lookupThreadCount.incrementAndGet());
+            thread.setDaemon(true);
+            return thread;
+        });
+        executor.allowCoreThreadTimeOut(true);
+        lookupExecutor = executor;
+    }
+
     protected void startInspection(Player player, Runnable inspection) {
         try {
             acquireInspection(player);
@@ -22,16 +53,31 @@ public abstract class BaseInspector {
             return;
         }
 
+        runLookup(player, () -> {
+            try {
+                inspection.run();
+            }
+            finally {
+                LookupThrottle.release(player.getName());
+            }
+        });
+    }
+
+    /**
+     * Runs a lookup on the shared inspector threads. The caller holds the player's throttle slot and the lookup
+     * releases it, so it is only released here when the lookup is rejected or discarded at shutdown.
+     */
+    public static void runLookup(Player player, Runnable lookup) {
+        ThreadPoolExecutor executor = lookupExecutor;
         try {
-            Thread thread = new Thread(() -> {
-                try {
-                    inspection.run();
-                }
-                finally {
-                    LookupThrottle.release(player.getName());
-                }
-            });
-            thread.start();
+            if (executor == null) {
+                throw new RejectedExecutionException();
+            }
+            executor.execute(new InspectorLookup(player.getName(), lookup));
+        }
+        catch (RejectedExecutionException e) {
+            LookupThrottle.release(player.getName());
+            Chat.sendMessage(player, Color.DARK_AQUA + "CoreProtect " + Color.WHITE + "- " + Phrase.build(Phrase.DATABASE_BUSY));
         }
         catch (RuntimeException | Error e) {
             LookupThrottle.release(player.getName());
@@ -39,6 +85,73 @@ public abstract class BaseInspector {
         }
     }
 
+    /**
+     * Lets queued lookups finish for a few seconds, then discards the rest and interrupts the running ones. Lookups
+     * that are still running after the interrupt are logged by thread and player.
+     */
+    public static void shutdown() {
+        ThreadPoolExecutor executor = lookupExecutor;
+        if (executor == null) {
+            return;
+        }
+
+        executor.shutdown();
+        try {
+            if (executor.awaitTermination(SHUTDOWN_WAIT_SECONDS, TimeUnit.SECONDS)) {
+                return;
+            }
+
+            discardQueued(executor.shutdownNow());
+            if (executor.awaitTermination(INTERRUPT_WAIT_SECONDS, TimeUnit.SECONDS)) {
+                return;
+            }
+        }
+        catch (InterruptedException e) {
+            discardQueued(executor.shutdownNow());
+            Thread.currentThread().interrupt();
+        }
+
+        for (InspectorLookup lookup : activeLookups) {
+            Thread thread = lookup.thread;
+            if (thread != null && thread.isAlive()) {
+                StackTraceElement[] stack = thread.getStackTrace();
+                String location = stack.length > 0 ? " at " + stack[0] : "";
+                Chat.console("Inspector lookup for " + lookup.playerName + " is still running on " + thread.getName() + location + " after disable.");
+            }
+        }
+    }
+
+    private static void discardQueued(List<Runnable> queued) {
+        for (Runnable runnable : queued) {
+            if (runnable instanceof InspectorLookup) {
+                LookupThrottle.release(((InspectorLookup) runnable).playerName);
+            }
+        }
+    }
+
+    private static final class InspectorLookup implements Runnable {
+        private final String playerName;
+        private final Runnable lookup;
+        private volatile Thread thread;
+
+        private InspectorLookup(String playerName, Runnable lookup) {
+            this.playerName = playerName;
+            this.lookup = lookup;
+        }
+
+        @Override
+        public void run() {
+            thread = Thread.currentThread();
+            activeLookups.add(this);
+            try {
+                lookup.run();
+            }
+            finally {
+                activeLookups.remove(this);
+            }
+        }
+    }
+
     private void acquireInspection(Player player) throws InspectionException {
         if (ConfigHandler.converterRunning) {
             throw new InspectionException(Color.DARK_AQUA + "CoreProtect " + Color.WHITE + "- " + Phrase.build(Phrase.UPGRADE_IN_PROGRESS));

src/main/java/net/coreprotect/services/PluginInitializationService.java

@@ -16,6 +16,7 @@ import net.coreprotect.consumer.Consumer;
 import net.coreprotect.language.Language;
 import net.coreprotect.language.Phrase;
 import net.coreprotect.listener.ListenerHandler;
+import net.coreprotect.listener.player.inspector.BaseInspector;
 import net.coreprotect.thread.CacheHandler;
 import net.coreprotect.thread.NetworkHandler;
 import net.coreprotect.thread.Scheduler;
@@ -31,6 +32,8 @@ import net.coreprotect.utility.ErrorReporter;
  */
 public class PluginInitializationService {
 
+    private static volatile Metrics metrics;
+
     private PluginInitializationService() {
         throw new IllegalStateException("Utility class");
     }
@@ -161,8 +164,10 @@ public class PluginInitializationService {
         TickTimeMonitor.initialize(plugin);
 
         // Start cache cleanup thread
-        Thread cacheCleanUpThread = new Thread(new CacheHandler());
-        cacheCleanUpThread.start();
+        CacheHandler.startThread();
+
+        // Start inspector lookup threads
+        BaseInspector.startLookups();
 
         Consumer.startConsumer();
         EntitySpawnTracking.initializeLoadedEntities();
@@ -177,10 +182,28 @@ public class PluginInitializationService {
      */
     private static void enableMetrics(JavaPlugin plugin) {
         try {
-            new Metrics(plugin, 2876);
+            metrics = new Metrics(plugin, 2876);
         }
         catch (Exception e) {
             // Failed to connect to bStats server or something else went wrong
         }
     }
+
+    /**
+     * Stops the bStats scheduler so its thread does not outlive a disable
+     */
+    public static void disableMetrics() {
+        Metrics current = metrics;
+        metrics = null;
+        if (current == null) {
+            return;
+        }
+
+        try {
+            current.shutdown();
+        }
+        catch (Exception e) {
+            ErrorReporter.report(e);
+        }
+    }
 }

src/main/java/net/coreprotect/services/ShutdownService.java

@@ -16,12 +16,15 @@ import net.coreprotect.language.Phrase;
 import net.coreprotect.listener.player.EntityInteractionListener;
 import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.listener.player.PlayerQuitListener;
+import net.coreprotect.listener.player.inspector.BaseInspector;
 import net.coreprotect.paper.PaperAdapter;
+import net.coreprotect.thread.CacheHandler;
 import net.coreprotect.utility.Chat;
 import net.coreprotect.utility.Extensions;
 import net.coreprotect.utility.EntitySpawnTracking;
 import net.coreprotect.utility.Teleport;
 import net.coreprotect.utility.ErrorReporter;
+import net.coreprotect.utility.VersionUtils;
 
 /**
  * Service responsible for handling plugin shutdown operations
@@ -45,7 +48,11 @@ public class ShutdownService {
     public static void safeShutdown(Plugin plugin) {
         try {
             Consumer.blockDatabaseReloadForShutdown();
+            if (ConfigHandler.worldeditEnabled) {
+                VersionUtils.unloadWorldEdit();
+            }
             Extensions.stopBackgroundService();
+            PluginInitializationService.disableMetrics();
 
             // Log disconnections of online players if server is stopping
             if (ConfigHandler.serverRunning && PaperAdapter.ADAPTER.isStopping(plugin.getServer())) {
@@ -87,6 +94,8 @@ public class ShutdownService {
                 ConfigHandler.shutdownDrainRunning = false;
             }
 
+            CacheHandler.stopThread(2000L);
+            BaseInspector.shutdown();
             ConfigHandler.performDisable();
             Chat.console(Phrase.build(Phrase.DISABLE_SUCCESS, "CoreProtect v" + plugin.getDescription().getVersion()));
         }

src/main/java/net/coreprotect/thread/CacheHandler.java

@@ -28,6 +28,31 @@ public class CacheHandler implements Runnable {
     public static ConcurrentHashMap<String, Object[]> redstoneCache = new ConcurrentHashMap<>(16, 0.75f, 2);
     public static ConcurrentHashMap<String, Object[]> fallingBlockSpawnCache = new ConcurrentHashMap<>(16, 0.75f, 2);
 
+    private static volatile Thread thread;
+
+    public static void startThread() {
+        Thread cacheThread = new Thread(new CacheHandler(), "CoreProtect-Cache");
+        cacheThread.setDaemon(true);
+        thread = cacheThread;
+        cacheThread.start();
+    }
+
+    public static void stopThread(long timeoutMillis) {
+        Thread cacheThread = thread;
+        thread = null;
+        if (cacheThread == null) {
+            return;
+        }
+
+        cacheThread.interrupt();
+        try {
+            cacheThread.join(timeoutMillis);
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
     public static String locationKey(Location location) {
         if (location == null || location.getWorld() == null) {
             return "";
@@ -152,6 +177,9 @@ public class CacheHandler implements Runnable {
                     }
                 }
             }
+            catch (InterruptedException e) {
+                return;
+            }
             catch (Exception e) {
                 ErrorReporter.report(e);
             }

src/main/java/net/coreprotect/worldedit/CoreProtectLogger.java

@@ -28,6 +28,7 @@ import com.sk89q.worldedit.world.block.BlockState;
 import com.sk89q.worldedit.world.block.BlockStateHolder;
 
 import net.coreprotect.config.Config;
+import net.coreprotect.config.ConfigHandler;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.utility.ItemUtils;
 
@@ -46,7 +47,7 @@ public class CoreProtectLogger extends AbstractDelegateExtent {
         Extent eventExtent = getExtent();
         org.bukkit.World world = BukkitAdapter.adapt(eventWorld);
         Config config = Config.getConfig(world);
-        if (!config.WORLDEDIT) {
+        if (!ConfigHandler.serverRunning || !config.WORLDEDIT) {
             return eventExtent.setBlock(position, block);
         }
 
@@ -84,7 +85,7 @@ public class CoreProtectLogger extends AbstractDelegateExtent {
     public int replaceBlocks(final Region region, final Mask mask, final Pattern pattern) throws MaxChangedBlocksException {
         Extent eventExtent = getExtent();
         org.bukkit.World world = BukkitAdapter.adapt(eventWorld);
-        if (!Config.getConfig(world).WORLDEDIT) {
+        if (!ConfigHandler.serverRunning || !Config.getConfig(world).WORLDEDIT) {
             return eventExtent.replaceBlocks(region, mask, pattern);
         }
 
@@ -102,7 +103,7 @@ public class CoreProtectLogger extends AbstractDelegateExtent {
     public int setBlocks(Region region, Pattern pattern) throws MaxChangedBlocksException {
         Extent eventExtent = getExtent();
         org.bukkit.World world = BukkitAdapter.adapt(eventWorld);
-        if (!Config.getConfig(world).WORLDEDIT) {
+        if (!ConfigHandler.serverRunning || !Config.getConfig(world).WORLDEDIT) {
             return eventExtent.setBlocks(region, pattern);
         }
 
@@ -113,7 +114,7 @@ public class CoreProtectLogger extends AbstractDelegateExtent {
     public int setBlocks(Set<BlockVector3> vset, Pattern pattern) {
         Extent eventExtent = getExtent();
         org.bukkit.World world = BukkitAdapter.adapt(eventWorld);
-        if (!Config.getConfig(world).WORLDEDIT) {
+        if (!ConfigHandler.serverRunning || !Config.getConfig(world).WORLDEDIT) {
             return eventExtent.setBlocks(vset, pattern);
         }
 

src/main/java/net/coreprotect/worldedit/FastAsyncWorldEditLogger.java

@@ -19,6 +19,7 @@ import com.sk89q.worldedit.world.block.BlockState;
 import com.sk89q.worldedit.world.block.BlockTypesCache;
 
 import net.coreprotect.config.Config;
+import net.coreprotect.config.ConfigHandler;
 
 final class FastAsyncWorldEditLogger implements IBatchProcessor {
     private static final ProcessorScope SCOPE = resolveScope();
@@ -41,7 +42,7 @@ final class FastAsyncWorldEditLogger implements IBatchProcessor {
     @Override
     public IChunkSet processSet(IChunk chunk, IChunkGet get, IChunkSet set) {
         Config config = Config.getConfig(world);
-        if (!config.WORLDEDIT) {
+        if (!ConfigHandler.serverRunning || !config.WORLDEDIT) {
             return set;
         }