Limit the total number of running threads

(Related #304)

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/FastLoginBukkit.java

@@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentMap;
 
 import org.bukkit.command.CommandSender;
 import org.bukkit.entity.Player;
+import org.bukkit.plugin.Plugin;
 import org.bukkit.plugin.PluginManager;
 import org.bukkit.plugin.java.JavaPlugin;
 import org.bukkit.plugin.messaging.PluginMessageRecipient;
@@ -68,6 +69,14 @@ public class FastLoginBukkit extends JavaPlugin implements PlatformPlugin<Comman
         }
 
         PluginManager pluginManager = getServer().getPluginManager();
+        Plugin protocolLib = pluginManager.getPlugin("ProtocolLib");
+        if (protocolLib != null && !protocolLib.isEnabled()) {
+            logger.warn("Dependency graph issue. ProtocolLib should be loaded first.");
+            logger.warn(getDescription().getSoftDepend().toString());
+            logger.warn(getDescription().getDepend().toString());
+            logger.warn(getDescription().getLoadBefore().toString());
+        }
+
         if (bungeeCord) {
             setServerStarted();
 

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/command/CrackedCommand.java

@@ -3,7 +3,6 @@ package com.github.games647.fastlogin.bukkit.command;
 import com.github.games647.fastlogin.bukkit.FastLoginBukkit;
 import com.github.games647.fastlogin.core.StoredProfile;
 
-import org.bukkit.Bukkit;
 import org.bukkit.command.Command;
 import org.bukkit.command.CommandSender;
 
@@ -44,7 +43,7 @@ public class CrackedCommand extends ToggleCommand {
 
                 profile.setPremium(false);
                 profile.setId(null);
-                Bukkit.getScheduler().runTaskAsynchronously(plugin, () -> {
+                plugin.getCore().getAsyncScheduler().runAsync(() -> {
                     plugin.getCore().getStorage().save(profile);
                 });
             } else {
@@ -76,7 +75,7 @@ public class CrackedCommand extends ToggleCommand {
             plugin.getCore().sendLocaleMessage("remove-premium", sender);
 
             profile.setPremium(false);
-            Bukkit.getScheduler().runTaskAsynchronously(plugin, () -> {
+            plugin.getCore().getAsyncScheduler().runAsync(() -> {
                 plugin.getCore().getStorage().save(profile);
             });
         }

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/command/PremiumCommand.java

@@ -5,7 +5,6 @@ import com.github.games647.fastlogin.core.StoredProfile;
 
 import java.util.UUID;
 
-import org.bukkit.Bukkit;
 import org.bukkit.command.Command;
 import org.bukkit.command.CommandSender;
 import org.bukkit.entity.Player;
@@ -56,7 +55,7 @@ public class PremiumCommand extends ToggleCommand {
             } else {
                 //todo: resolve uuid
                 profile.setPremium(true);
-                Bukkit.getScheduler().runTaskAsynchronously(plugin, () -> {
+                plugin.getCore().getAsyncScheduler().runAsync(() -> {
                     plugin.getCore().getStorage().save(profile);
                 });
 
@@ -85,7 +84,7 @@ public class PremiumCommand extends ToggleCommand {
         } else {
             //todo: resolve uuid
             profile.setPremium(true);
-            Bukkit.getScheduler().runTaskAsynchronously(plugin, () -> {
+            plugin.getCore().getAsyncScheduler().runAsync(() -> {
                 plugin.getCore().getStorage().save(profile);
             });
 

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/listener/protocollib/ProtocolLibListener.java

@@ -11,7 +11,6 @@ import com.github.games647.fastlogin.bukkit.FastLoginBukkit;
 import java.security.KeyPair;
 import java.security.SecureRandom;
 
-import org.bukkit.Bukkit;
 import org.bukkit.entity.Player;
 
 import static com.comphenix.protocol.PacketType.Login.Client.ENCRYPTION_BEGIN;
@@ -67,7 +66,7 @@ public class ProtocolLibListener extends PacketAdapter {
 
         packetEvent.getAsyncMarker().incrementProcessingDelay();
         Runnable verifyTask = new VerifyResponseTask(plugin, packetEvent, sender, sharedSecret, keyPair);
-        Bukkit.getScheduler().runTaskAsynchronously(plugin, verifyTask);
+        plugin.getCore().getAsyncScheduler().runAsync(verifyTask);
     }
 
     private void onLogin(PacketEvent packetEvent, Player player) {
@@ -85,6 +84,6 @@ public class ProtocolLibListener extends PacketAdapter {
 
         packetEvent.getAsyncMarker().incrementProcessingDelay();
         Runnable nameCheckTask = new NameCheckTask(plugin, packetEvent, random, player, username, keyPair.getPublic());
-        Bukkit.getScheduler().runTaskAsynchronously(plugin, nameCheckTask);
+        plugin.getCore().getAsyncScheduler().runAsync(nameCheckTask);
     }
 }

bukkit/src/main/resources/plugin.yml

@@ -11,21 +11,14 @@ description: |
 website: ${project.url}
 dev-url: ${project.url}
 
+# Load the plugin as early as possible to inject it for all players
+load: STARTUP
+
 # This plugin don't have to be transformed for compatibility with Minecraft >= 1.13
 api-version: '1.13'
 
-softdepend:
-    # We depend either ProtocolLib or ProtocolSupport
-    - ProtocolSupport
-    - ProtocolLib
-    - PlaceholderAPI
-    # Auth plugins
-    - AuthMe
-    - LoginSecurity
-    - xAuth
-    - LogIt
-    - UltraAuth
-    - CrazyLogin
+# We depend either ProtocolLib or ProtocolSupport
+depend: [ProtocolLib]
 
 commands:
     ${project.parent.name}:

bungee/src/main/java/com/github/games647/fastlogin/bungee/FastLoginBungee.java

@@ -117,7 +117,7 @@ public class FastLoginBungee extends Plugin implements PlatformPlugin<CommandSen
     @SuppressWarnings("deprecation")
     public ThreadFactory getThreadFactory() {
         return new ThreadFactoryBuilder()
-                .setNameFormat(getName() + " Database Pool Thread #%1$d")
+                .setNameFormat(getName() + " Pool Thread #%1$d")
                 //Hikari create daemons by default
                 .setDaemon(true)
                 .setThreadFactory(new GroupedThreadFactory(this, getName()))

bungee/src/main/java/com/github/games647/fastlogin/bungee/listener/ConnectListener.java

@@ -10,7 +10,6 @@ import com.github.games647.fastlogin.core.shared.LoginSession;
 import java.lang.reflect.Field;
 import java.util.UUID;
 
-import net.md_5.bungee.api.ProxyServer;
 import net.md_5.bungee.api.connection.PendingConnection;
 import net.md_5.bungee.api.connection.ProxiedPlayer;
 import net.md_5.bungee.api.connection.Server;
@@ -48,7 +47,7 @@ public class ConnectListener implements Listener {
 
         PendingConnection connection = preLoginEvent.getConnection();
         Runnable asyncPremiumCheck = new AsyncPremiumCheck(plugin, preLoginEvent, connection);
-        ProxyServer.getInstance().getScheduler().runAsync(plugin, asyncPremiumCheck);
+        plugin.getCore().getAsyncScheduler().runAsync(asyncPremiumCheck);
     }
 
     @EventHandler(priority = EventPriority.LOWEST)
@@ -101,7 +100,7 @@ public class ConnectListener implements Listener {
         Server server = serverConnectedEvent.getServer();
 
         Runnable loginTask = new ForceLoginTask(plugin.getCore(), player, server);
-        ProxyServer.getInstance().getScheduler().runAsync(plugin, loginTask);
+        plugin.getCore().getAsyncScheduler().runAsync(loginTask);
     }
 
     @EventHandler

bungee/src/main/java/com/github/games647/fastlogin/bungee/listener/PluginMessageListener.java

@@ -14,7 +14,6 @@ import com.google.common.io.ByteStreams;
 import java.util.Arrays;
 
 import net.md_5.bungee.api.CommandSender;
-import net.md_5.bungee.api.ProxyServer;
 import net.md_5.bungee.api.chat.TextComponent;
 import net.md_5.bungee.api.connection.ProxiedPlayer;
 import net.md_5.bungee.api.connection.Server;
@@ -56,7 +55,7 @@ public class PluginMessageListener implements Listener {
         byte[] data = Arrays.copyOf(pluginMessageEvent.getData(), pluginMessageEvent.getData().length);
         ProxiedPlayer forPlayer = (ProxiedPlayer) pluginMessageEvent.getReceiver();
 
-        ProxyServer.getInstance().getScheduler().runAsync(plugin, () -> readMessage(forPlayer, channel, data));
+        plugin.getCore().getAsyncScheduler().runAsync(() -> readMessage(forPlayer, channel, data));
     }
 
     private void readMessage(ProxiedPlayer forPlayer, String channel, byte[] data) {
@@ -82,10 +81,10 @@ public class PluginMessageListener implements Listener {
 
                 core.getPendingConfirms().remove(forPlayer.getUniqueId());
                 Runnable task = new AsyncToggleMessage(core, forPlayer, playerName, true, isSourceInvoker);
-                ProxyServer.getInstance().getScheduler().runAsync(plugin, task);
+                plugin.getCore().getAsyncScheduler().runAsync(task);
             } else {
                 Runnable task = new AsyncToggleMessage(core, forPlayer, playerName, false, isSourceInvoker);
-                ProxyServer.getInstance().getScheduler().runAsync(plugin, task);
+                plugin.getCore().getAsyncScheduler().runAsync(task);
             }
         }
     }

core/src/main/java/com/github/games647/fastlogin/core/AsyncScheduler.java

@@ -0,0 +1,40 @@
+package com.github.games647.fastlogin.core;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This limits the number of threads that are used at maximum. Thread creation can be very heavy for the CPU and
+ * context switching between threads too. However we need many threads for blocking HTTP and database calls.
+ * Nevertheless this number can be further limited, because the number of actually working database threads
+ * is limited by the size of our database pool. The goal is to separate concerns into processing and blocking only
+ * threads.
+ */
+public class AsyncScheduler {
+
+    // 30 threads are still too many - the optimal solution is to separate into processing and blocking threads
+    // where processing threads could only be max number of cores while blocking threads could be minimized using
+    // non-blocking I/O and a single event executor
+    private final ThreadPoolExecutor executorService;
+
+    public AsyncScheduler(ThreadFactory threadFactory) {
+        executorService = new ThreadPoolExecutor(5, 30,
+                60L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<>(1024), threadFactory);
+    }
+
+    public void runAsync(Runnable task) {
+        executorService.execute(task);
+    }
+
+    public void shutdown() {
+        executorService.shutdown();
+        try {
+            executorService.awaitTermination(1, TimeUnit.MINUTES);
+        } catch (InterruptedException interruptEx) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}

core/src/main/java/com/github/games647/fastlogin/core/AuthStorage.java

@@ -72,6 +72,9 @@ public class AuthStorage {
     }
 
     public void createTables() throws SQLException {
+        // choose surrogate PK(ID), because UUID can be null for offline players
+        // if UUID is always Premium UUID we would have to update offline player entries on insert
+        // name cannot be PK, because it can be changed for premium players
         String createDataStmt = "CREATE TABLE IF NOT EXISTS `" + PREMIUM_TABLE + "` ("
                 + "`UserID` INTEGER PRIMARY KEY AUTO_INCREMENT, "
                 + "`UUID` CHAR(36), "
@@ -87,7 +90,7 @@ public class AuthStorage {
             createDataStmt = createDataStmt.replace("AUTO_INCREMENT", "AUTOINCREMENT");
         }
 
-        //todo: add uuid index usage
+        //todo: add unique uuid index usage
         try (Connection con = dataSource.getConnection();
              Statement createStmt = con.createStatement()) {
             createStmt.executeUpdate(createDataStmt);

core/src/main/java/com/github/games647/fastlogin/core/shared/FastLoginCore.java

@@ -4,6 +4,7 @@ import com.github.games647.craftapi.resolver.MojangResolver;
 import com.github.games647.craftapi.resolver.http.RotatingProxySelector;
 import com.github.games647.fastlogin.core.AuthStorage;
 import com.github.games647.fastlogin.core.CommonUtil;
+import com.github.games647.fastlogin.core.AsyncScheduler;
 import com.github.games647.fastlogin.core.hooks.AuthPlugin;
 import com.github.games647.fastlogin.core.hooks.DefaultPasswordGenerator;
 import com.github.games647.fastlogin.core.hooks.PasswordGenerator;
@@ -51,6 +52,7 @@ public class FastLoginCore<P extends C, C, T extends PlatformPlugin<C>> {
     private final T plugin;
 
     private final MojangResolver resolver = new MojangResolver();
+    private final AsyncScheduler scheduler;
 
     private Configuration config;
     private AuthStorage storage;
@@ -59,6 +61,7 @@ public class FastLoginCore<P extends C, C, T extends PlatformPlugin<C>> {
 
     public FastLoginCore(T plugin) {
         this.plugin = plugin;
+        this.scheduler = new AsyncScheduler(plugin.getThreadFactory());
     }
 
     public void load() {
@@ -239,7 +242,14 @@ public class FastLoginCore<P extends C, C, T extends PlatformPlugin<C>> {
         }
     }
 
+    public AsyncScheduler getAsyncScheduler() {
+        return scheduler;
+    }
+
     public void close() {
+        plugin.getLog().info("Safely shutting down scheduler. This could take up to one minute.");
+        scheduler.shutdown();
+
         if (storage != null) {
             storage.close();
         }

core/src/main/java/com/github/games647/fastlogin/core/shared/PlatformPlugin.java

@@ -1,5 +1,7 @@
 package com.github.games647.fastlogin.core.shared;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
 import java.nio.file.Path;
 import java.util.concurrent.ThreadFactory;
 
@@ -22,6 +24,11 @@ public interface PlatformPlugin<C> {
     }
 
     default ThreadFactory getThreadFactory() {
-        return null;
+        return new ThreadFactoryBuilder()
+                .setNameFormat(getName() + " Pool Thread #%1$d")
+                // Hikari create daemons by default and we could daemon threads for our own scheduler too
+                // because we safely shutdown
+                .setDaemon(true)
+                .build();
     }
 }