Compare commits

...

6 Commits

Author SHA1 Message Date
games647
18d2ba5095 Disable restore session cancel 2020-03-20 13:51:12 +01:00
games647
65469ed579 Make scheduler platform dependent 2020-03-20 13:44:58 +01:00
games647
8ecb5657d3 Drop nullable annotation usage to compile
Thanks CI
2020-03-14 19:06:05 +01:00
games647
b0cf6e39c7 Drop forced dependency 2020-03-14 19:01:09 +01:00
games647
88a526b5bf Drop support for Minecraft 1.7 2020-03-14 18:51:27 +01:00
games647
57b84509da Limit the total number of running threads
(Related #304)
2020-03-14 18:20:34 +01:00
16 changed files with 161 additions and 54 deletions

View File

@@ -51,7 +51,7 @@ https://ci.codemc.org/job/Games647/job/FastLogin/changes
* Plugin:
* [ProtocolLib](https://www.spigotmc.org/resources/protocollib.1997/) or
* [ProtocolSupport](https://www.spigotmc.org/resources/protocolsupport.7201/)
* [Spigot](https://www.spigotmc.org) 1.7.10+
* [Spigot](https://www.spigotmc.org) 1.8.8+
* Java 8+
* Run Spigot and/or BungeeCord/Waterfall in offline mode (see server.properties or config.yml)
* An auth plugin. Supported plugins

View File

@@ -0,0 +1,27 @@
package com.github.games647.fastlogin.bukkit;
import com.github.games647.fastlogin.core.AsyncScheduler;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadFactory;
import org.bukkit.Bukkit;
import org.bukkit.plugin.Plugin;
import org.slf4j.Logger;
public class BukkitScheduler extends AsyncScheduler {
private final Plugin plugin;
private final Executor syncExecutor;
public BukkitScheduler(Plugin plugin, Logger logger, ThreadFactory threadFactory) {
super(logger, threadFactory);
this.plugin = plugin;
syncExecutor = r -> Bukkit.getScheduler().runTask(plugin, r);
}
public Executor getSyncExecutor() {
return syncExecutor;
}
}

View File

@@ -41,17 +41,24 @@ public class FastLoginBukkit extends JavaPlugin implements PlatformPlugin<Comman
//1 minutes should be enough as a timeout for bad internet connection (Server, Client and Mojang)
private final ConcurrentMap<String, BukkitLoginSession> loginSession = CommonUtil.buildCache(1, -1);
private final Logger logger = CommonUtil.createLoggerFromJDK(getLogger());
private final Map<UUID, PremiumStatus> premiumPlayers = new ConcurrentHashMap<>();
private final Logger logger;
private boolean serverStarted;
private boolean bungeeCord;
private final BukkitScheduler scheduler;
private FastLoginCore<Player, CommandSender, FastLoginBukkit> core;
public FastLoginBukkit() {
this.logger = CommonUtil.createLoggerFromJDK(getLogger());
this.scheduler = new BukkitScheduler(this, logger, getThreadFactory());
}
@Override
public void onEnable() {
core = new FastLoginCore<>(this);
core.load();
try {
bungeeCord = Class.forName("org.spigotmc.SpigotConfig").getDeclaredField("bungee").getBoolean(null);
} catch (ClassNotFoundException notFoundEx) {
@@ -72,7 +79,7 @@ public class FastLoginBukkit extends JavaPlugin implements PlatformPlugin<Comman
setServerStarted();
// check for incoming messages from the bungeecord version of this plugin
String forceChannel = new NamespaceKey(getName(), LoginActionMessage.FORCE_CHANNEL).getCombinedName();
String forceChannel = NamespaceKey.getCombined(getName(), LoginActionMessage.FORCE_CHANNEL);
getServer().getMessenger().registerIncomingPluginChannel(this, forceChannel, new BungeeListener(this));
// outgoing
@@ -195,6 +202,11 @@ public class FastLoginBukkit extends JavaPlugin implements PlatformPlugin<Comman
return logger;
}
@Override
public BukkitScheduler getScheduler() {
return scheduler;
}
@Override
public void sendMessage(CommandSender receiver, String message) {
receiver.sendMessage(message);

View File

@@ -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.getScheduler().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.getScheduler().runAsync(() -> {
plugin.getCore().getStorage().save(profile);
});
}

View File

@@ -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.getScheduler().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.getScheduler().runAsync(() -> {
plugin.getCore().getStorage().save(profile);
});

View File

@@ -1,15 +1,11 @@
package com.github.games647.fastlogin.bukkit.hook;
import com.github.games647.fastlogin.bukkit.BukkitLoginSession;
import com.github.games647.fastlogin.bukkit.FastLoginBukkit;
import com.github.games647.fastlogin.core.hooks.AuthPlugin;
import fr.xephi.authme.api.v3.AuthMeApi;
import fr.xephi.authme.events.RestoreSessionEvent;
import org.bukkit.entity.Player;
import org.bukkit.event.EventHandler;
import org.bukkit.event.EventPriority;
import org.bukkit.event.Listener;
/**
@@ -29,17 +25,6 @@ public class AuthMeHook implements AuthPlugin<Player>, Listener {
this.plugin = plugin;
}
@EventHandler(priority = EventPriority.HIGHEST, ignoreCancelled = true)
public void onSessionRestore(RestoreSessionEvent restoreSessionEvent) {
Player player = restoreSessionEvent.getPlayer();
String id = '/' + player.getAddress().getAddress().getHostAddress() + ':' + player.getAddress().getPort();
BukkitLoginSession session = plugin.getLoginSessions().get(id);
if (session != null && session.isVerified()) {
restoreSessionEvent.setCancelled(true);
}
}
@Override
public boolean forceLogin(Player player) {
if (AuthMeApi.getInstance().isAuthenticated(player)) {

View File

@@ -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.getScheduler().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.getScheduler().runAsync(nameCheckTask);
}
}

View File

@@ -3,6 +3,7 @@ package com.github.games647.fastlogin.bungee;
import com.github.games647.fastlogin.bungee.hook.BungeeAuthHook;
import com.github.games647.fastlogin.bungee.listener.ConnectListener;
import com.github.games647.fastlogin.bungee.listener.PluginMessageListener;
import com.github.games647.fastlogin.core.AsyncScheduler;
import com.github.games647.fastlogin.core.CommonUtil;
import com.github.games647.fastlogin.core.message.ChangePremiumMessage;
import com.github.games647.fastlogin.core.message.ChannelMessage;
@@ -37,11 +38,13 @@ public class FastLoginBungee extends Plugin implements PlatformPlugin<CommandSen
private final ConcurrentMap<PendingConnection, BungeeLoginSession> session = new MapMaker().weakKeys().makeMap();
private FastLoginCore<ProxiedPlayer, CommandSender, FastLoginBungee> core;
private AsyncScheduler scheduler;
private Logger logger;
@Override
public void onEnable() {
logger = CommonUtil.createLoggerFromJDK(getLogger());
scheduler = new AsyncScheduler(logger, getThreadFactory());
core = new FastLoginCore<>(this);
core.load();
@@ -54,8 +57,8 @@ public class FastLoginBungee extends Plugin implements PlatformPlugin<CommandSen
getProxy().getPluginManager().registerListener(this, new PluginMessageListener(this));
//this is required to listen to incoming messages from the server
getProxy().registerChannel(new NamespaceKey(getName(), ChangePremiumMessage.CHANGE_CHANNEL).getCombinedName());
getProxy().registerChannel(new NamespaceKey(getName(), SuccessMessage.SUCCESS_CHANNEL).getCombinedName());
getProxy().registerChannel(NamespaceKey.getCombined(getName(), ChangePremiumMessage.CHANGE_CHANNEL));
getProxy().registerChannel(NamespaceKey.getCombined(getName(), SuccessMessage.SUCCESS_CHANNEL));
registerHook();
}
@@ -117,10 +120,15 @@ 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()))
.build();
}
@Override
public AsyncScheduler getScheduler() {
return scheduler;
}
}

View File

@@ -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.getScheduler().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.getScheduler().runAsync(loginTask);
}
@EventHandler

View File

@@ -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.getScheduler().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.getScheduler().runAsync(task);
} else {
Runnable task = new AsyncToggleMessage(core, forPlayer, playerName, false, isSourceInvoker);
ProxyServer.getInstance().getScheduler().runAsync(plugin, task);
plugin.getScheduler().runAsync(task);
}
}
}

View File

@@ -45,13 +45,6 @@
<version>1.7.30</version>
</dependency>
<!--GSON is not at the right position for Minecraft 1.7-->
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.2.4</version>
</dependency>
<!-- snakeyaml is present in Bungee, Spigot, Cauldron and so we could use this independent implementation -->
<dependency>
<groupId>net.md-5</groupId>
@@ -69,17 +62,21 @@
<dependency>
<groupId>com.github.games647</groupId>
<artifactId>craftapi</artifactId>
<version>0.3</version>
<version>0.4</version>
</dependency>
<!-- APIs we can use because they are available in all platforms (Spigot, Bungee, Cauldron) -->
<!-- APIs we can use because they are available in all platforms (Spigot, Bungee) -->
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<!-- The Uranium project (fork of Cauldron) uses 17.0 like Spigot 1.8 as experimental feature -->
<!-- Project url: https://github.com/UraniumMC/Uranium -->
<version>10.0.1</version>
<version>17.0</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.2.4</version>
</dependency>
</dependencies>
</project>

View File

@@ -0,0 +1,69 @@
package com.github.games647.fastlogin.core;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
/**
* 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 {
private static final int MAX_CAPACITY = 1024;
//todo: single thread for delaying and scheduling tasks
private final Logger logger;
// 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 ExecutorService processingPool;
/*
private final ExecutorService databaseExecutor = new ThreadPoolExecutor(1, 10,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(MAX_CAPACITY));
*/
public AsyncScheduler(Logger logger, ThreadFactory threadFactory) {
this.logger = logger;
processingPool = new ThreadPoolExecutor(6, 32,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(MAX_CAPACITY), threadFactory);
}
/*
public <R> CompletableFuture<R> runDatabaseTask(Supplier<R> databaseTask) {
return CompletableFuture.supplyAsync(databaseTask, databaseExecutor)
.exceptionally(error -> {
logger.warn("Error occurred on thread pool", error);
return null;
})
// change context to the processing pool
.thenApplyAsync(r -> r, processingPool);
}
*/
public CompletableFuture<Void> runAsync(Runnable task) {
return CompletableFuture.runAsync(task, processingPool).exceptionally(error -> {
logger.warn("Error occurred on thread pool", error);
return null;
});
}
public void shutdown() {
MoreExecutors.shutdownAndAwaitTermination(processingPool, 1, TimeUnit.MINUTES);
//MoreExecutors.shutdownAndAwaitTermination(databaseExecutor, 1, TimeUnit.MINUTES);
}
}

View File

@@ -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);

View File

@@ -6,8 +6,6 @@ import java.time.Instant;
import java.util.Optional;
import java.util.UUID;
import javax.annotation.Nullable;
public class StoredProfile extends Profile {
private long rowId;
@@ -45,7 +43,7 @@ public class StoredProfile extends Profile {
this.rowId = generatedId;
}
@Nullable
// can be null
public synchronized UUID getId() {
return id;
}

View File

@@ -240,6 +240,9 @@ public class FastLoginCore<P extends C, C, T extends PlatformPlugin<C>> {
}
public void close() {
plugin.getLog().info("Safely shutting down scheduler. This could take up to one minute.");
plugin.getScheduler().shutdown();
if (storage != null) {
storage.close();
}

View File

@@ -1,5 +1,8 @@
package com.github.games647.fastlogin.core.shared;
import com.github.games647.fastlogin.core.AsyncScheduler;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.nio.file.Path;
import java.util.concurrent.ThreadFactory;
@@ -15,6 +18,8 @@ public interface PlatformPlugin<C> {
void sendMessage(C receiver, String message);
AsyncScheduler getScheduler();
default void sendMultiLineMessage(C receiver, String message) {
for (String line : message.split("%nl%")) {
sendMessage(receiver, line);
@@ -22,6 +27,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();
}
}