From c11af0da512339771e9169dd702c397250d30c7f Mon Sep 17 00:00:00 2001 From: anthorob Date: Tue, 21 Sep 2021 14:03:50 -0400 Subject: [PATCH] Updated threadmanager and refactored code --- src/main/java/netneutrality/MusicBot.java | 2 +- .../java/netneutrality/ThreadManager.java | 96 ++++++++++++------- .../netneutrality/db/AsyncMySqlAction.java | 22 ++--- .../java/netneutrality/db/DBController.java | 3 +- .../db/controllers/DBBlacklist.java | 3 +- .../db/controllers/DBLogging.java | 5 +- .../netneutrality/handler/ConsoleHandler.java | 2 + .../http/server/ActionHandler.java | 28 ++---- .../netneutrality/http/server/WebApi.java | 10 +- .../netneutrality/music/MusicManager.java | 8 +- .../netneutrality/utils/ChannelFunction.java | 2 +- 11 files changed, 103 insertions(+), 78 deletions(-) diff --git a/src/main/java/netneutrality/MusicBot.java b/src/main/java/netneutrality/MusicBot.java index db95900..c1559d2 100644 --- a/src/main/java/netneutrality/MusicBot.java +++ b/src/main/java/netneutrality/MusicBot.java @@ -42,7 +42,7 @@ public class MusicBot { jda.getPresence().setActivity(Activity.watching("me load...")); jda.addEventListener(new NewCommandListener(), new EditHandler()); - WebApi webserver = new WebApi(); + new WebApi(); jda.getPresence().setActivity(Activity.watching("you")); diff --git a/src/main/java/netneutrality/ThreadManager.java b/src/main/java/netneutrality/ThreadManager.java index 1c24f4b..7fb860a 100644 --- a/src/main/java/netneutrality/ThreadManager.java +++ b/src/main/java/netneutrality/ThreadManager.java @@ -1,68 +1,92 @@ package netneutrality; -/* - * Copyright (c) anthorob - * - * This file is closed-source. It should not be distributed or copied in any way or form - */ - import netneutrality.thread.BaseThread; -import netneutrality.utils.Settings; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +import java.util.Map; +import java.util.concurrent.*; public class ThreadManager { - private static ThreadManager INSTANCE; + public enum ExecutorType { + SINGLE, SCHEDULED, FIXED + } + + private static final Map executors = new ConcurrentHashMap<>(); private final static Logger logger = LoggerFactory.getLogger(ThreadManager.class); - public static final ExecutorService executor = Executors.newSingleThreadExecutor(new BaseThread("executor")); - public static final ExecutorService dBExecutor = Executors.newSingleThreadExecutor(new BaseThread("db-action")); - public static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5, new BaseThread("scheduled-action")); - - public ThreadManager() { + public static ExecutorService getExecutor(String name) { + return executors.computeIfAbsent(name, s -> Executors.newSingleThreadExecutor(new BaseThread(name))); } - public static ThreadManager getInstance() { - if(INSTANCE == null) { - INSTANCE = new ThreadManager(); + public static ExecutorService getExecutor(String name, ExecutorType type) { + ExecutorService executor; + + switch (type) { + case FIXED: + executor = executors.computeIfAbsent(name, s -> Executors.newFixedThreadPool(5, new BaseThread(name))); + break; + case SCHEDULED: + executor = executors.computeIfAbsent(name, s -> Executors.newScheduledThreadPool(5, new BaseThread(name))); + break; + default: + executor = executors.computeIfAbsent(name, s -> Executors.newSingleThreadExecutor(new BaseThread(name))); } - return INSTANCE; + return executor; + } + + public static ExecutorService getExecutor(String name, ExecutorType type, Integer poolSize) { + ExecutorService executor; + + switch (type) { + case FIXED: + executor = executors.computeIfAbsent(name, s -> Executors.newFixedThreadPool(poolSize, new BaseThread(name))); + break; + case SCHEDULED: + executor = executors.computeIfAbsent(name, s -> Executors.newScheduledThreadPool(poolSize, new BaseThread(name))); + break; + default: + executor = executors.computeIfAbsent(name, s -> Executors.newSingleThreadExecutor(new BaseThread(name))); + } + + return executor; + } + + + public static Future dispatch(ExecutorService executor, Runnable task) { + return executor.submit(task); + } + + public static Future dispatch(ExecutorService executor, Callable task) { + return executor.submit(task); } public static void shutdownExecutor() { - logger.info("[Executor] Shutting down..."); + logger.debug("[Executor] Shutting down..."); try { - executor.shutdown(); - scheduler.shutdown(); - executor.awaitTermination(5, TimeUnit.SECONDS); - scheduler.awaitTermination(5, TimeUnit.SECONDS); + for (ExecutorService ex : executors.values()) { + ex.shutdown(); + ex.awaitTermination(5, TimeUnit.SECONDS); + } } catch (InterruptedException e) { logger.warn("[Executor] Running tasks interrupted"); } finally { - if (!executor.isTerminated()) { - logger.warn("[Executor] Forcing shutdown of running tasks.."); - executor.shutdownNow(); - } - - if (!scheduler.isTerminated()) { - logger.warn("[Executor] Forcing shutdown of running tasks.."); - scheduler.shutdownNow(); - } + executors.forEach((name, ex) -> { + if (!ex.isTerminated()) { + logger.warn("[Executor: " + name + "] Forcing shutdown of running tasks.."); + ex.shutdownNow(); + } + }); } - logger.info("[Executor] Shutdown finished!"); + logger.debug("[Executor] Shutdown finished!"); } diff --git a/src/main/java/netneutrality/db/AsyncMySqlAction.java b/src/main/java/netneutrality/db/AsyncMySqlAction.java index 63b0d99..2e84098 100644 --- a/src/main/java/netneutrality/db/AsyncMySqlAction.java +++ b/src/main/java/netneutrality/db/AsyncMySqlAction.java @@ -8,17 +8,15 @@ package netneutrality.db; */ import netneutrality.ThreadManager; -import netneutrality.db.entities.EntityInterface; import netneutrality.db.models.MTableModel; import java.sql.ResultSet; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; +import java.util.concurrent.*; import java.util.function.Consumer; public class AsyncMySqlAction { + private final ExecutorService executor = ThreadManager.getExecutor("db-action", ThreadManager.ExecutorType.SINGLE); private final String sqlString; private final Object[] params; private final MysqlDatabase.CommandType type; @@ -39,24 +37,24 @@ public class AsyncMySqlAction { public void async(Consumer success) { switch (type) { case INSERT: - ThreadManager.dBExecutor.submit(() -> { + ThreadManager.dispatch(executor, () -> { db.insert(this.sqlString, this.params); }); break; case UPDATE: - ThreadManager.dBExecutor.submit(() -> { + ThreadManager.dispatch(executor, () -> { db.update(this.sqlString, this.params); }); break; case DELETE: - ThreadManager.dBExecutor.submit(() -> { + ThreadManager.dispatch(executor, () -> { db.delete(this.sqlString, this.params); }); break; case SELECT: - ThreadManager.dBExecutor.submit(() -> { + ThreadManager.dispatch(executor, () -> { ResultSet result = db.prepareAndExecute(this.sqlString, this.params); MTableModel table = MysqlDatabase.convertToTableModel(result); @@ -70,17 +68,17 @@ public class AsyncMySqlAction { try { switch (type) { case INSERT: - ThreadManager.dBExecutor.submit(() -> { + ThreadManager.dispatch(executor, () -> { db.insert(this.sqlString, this.params); }).get(); case UPDATE: - ThreadManager.dBExecutor.submit(() -> { + ThreadManager.dispatch(executor, () -> { db.update(this.sqlString, this.params); }).get(); case DELETE: - ThreadManager.dBExecutor.submit(() -> { + ThreadManager.dispatch(executor, () -> { db.delete(this.sqlString, this.params); }).get(); @@ -92,7 +90,7 @@ public class AsyncMySqlAction { return table; }; - Future future = ThreadManager.dBExecutor.submit(task); + Future future = ThreadManager.dispatch(executor, task); return future.get(); } } catch (InterruptedException | ExecutionException e) { diff --git a/src/main/java/netneutrality/db/DBController.java b/src/main/java/netneutrality/db/DBController.java index 6be429e..e572608 100644 --- a/src/main/java/netneutrality/db/DBController.java +++ b/src/main/java/netneutrality/db/DBController.java @@ -5,6 +5,7 @@ import java.lang.reflect.Modifier; import java.sql.SQLException; import java.util.HashMap; import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import netneutrality.MusicBot; @@ -70,7 +71,7 @@ public class DBController { if (controllers.isEmpty()) loadControllers(); - ThreadManager.scheduler.scheduleWithFixedDelay(() -> { + ((ScheduledExecutorService) ThreadManager.getExecutor("db-keepalive", ThreadManager.ExecutorType.SCHEDULED, 5)).scheduleWithFixedDelay(() -> { DBController.get().query("SELECT 1;"); }, 0, 120, TimeUnit.SECONDS); } diff --git a/src/main/java/netneutrality/db/controllers/DBBlacklist.java b/src/main/java/netneutrality/db/controllers/DBBlacklist.java index f2b9370..352f85d 100644 --- a/src/main/java/netneutrality/db/controllers/DBBlacklist.java +++ b/src/main/java/netneutrality/db/controllers/DBBlacklist.java @@ -80,7 +80,8 @@ public class DBBlacklist extends AbstractController { @Deprecated public void getBlacklistForWord(String word, String serverId, Consumer success) { - ThreadManager.executor.submit(() -> { + + ThreadManager.getExecutor("db-action").submit(() -> { MBlacklist blacklisted = null; if (isAlreadyBlacklisted(word, serverId)) { diff --git a/src/main/java/netneutrality/db/controllers/DBLogging.java b/src/main/java/netneutrality/db/controllers/DBLogging.java index efc4c15..41805f4 100644 --- a/src/main/java/netneutrality/db/controllers/DBLogging.java +++ b/src/main/java/netneutrality/db/controllers/DBLogging.java @@ -22,7 +22,7 @@ import java.util.function.Consumer; public class DBLogging extends AbstractController { public void logGuild(Guild g) { - ThreadManager.executor.submit(() -> { + ThreadManager.getExecutor("db-action").submit(() -> { if (!database.exists("guilds", "guild_id", g.getId())) { database.insert("INSERT INTO guilds VALUES(?, ?, ?)", g.getId(), g.getName(), g.getOwner().getUser().getName()); } @@ -30,7 +30,8 @@ public class DBLogging extends AbstractController { } public void getLoggingChannel(String serverId, Consumer success) { - ThreadManager.executor.submit(() -> { + + ThreadManager.getExecutor("db-action").submit(() -> { TextChannel txt = null; ResultSet rs = database.prepareAndExecute("SELECT channelId FROM logging WHERE serverId = ?", serverId); diff --git a/src/main/java/netneutrality/handler/ConsoleHandler.java b/src/main/java/netneutrality/handler/ConsoleHandler.java index 3988453..6a489e0 100644 --- a/src/main/java/netneutrality/handler/ConsoleHandler.java +++ b/src/main/java/netneutrality/handler/ConsoleHandler.java @@ -7,8 +7,10 @@ import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; +//TODO: To be replaced by HTTP API actions public class ConsoleHandler { + private final static Logger logger = LoggerFactory.getLogger(ConsoleHandler.class); public ConsoleHandler() { logger.info("Console input ready!"); diff --git a/src/main/java/netneutrality/http/server/ActionHandler.java b/src/main/java/netneutrality/http/server/ActionHandler.java index 83daaec..c36f4dc 100644 --- a/src/main/java/netneutrality/http/server/ActionHandler.java +++ b/src/main/java/netneutrality/http/server/ActionHandler.java @@ -15,28 +15,18 @@ import java.io.OutputStream; public class ActionHandler implements HttpHandler { @Override - public void handle(HttpExchange exchange) { + public void handle(HttpExchange exchange) throws IOException { - try { - String path = exchange.getRequestURI().getRawPath(); + String path = exchange.getRequestURI().getRawPath(); - if (path.equals("/actions/shutdown")) { - LogManager.getLogger().info("Request to shutdown received"); - exchange.sendResponseHeaders(200, "Shutting down..".length()); - OutputStream os = exchange.getResponseBody(); - os.write("Shutting down..".getBytes()); - os.close(); - //MusicBot.shutdown(); - return; - } - - exchange.sendResponseHeaders(200, path.length()); - OutputStream os = exchange.getResponseBody(); - os.write(path.getBytes()); - os.close(); - } catch (Exception ex) { - ex.printStackTrace(); + if (path.equals("/actions/shutdown")) { + LogManager.getLogger(MusicBot.class).info("Request to shutdown received"); + WebApi.sendStringResponse(exchange, "Shutting down.."); + MusicBot.shutdown(); + return; } + WebApi.sendStringResponse(exchange, path); + } } diff --git a/src/main/java/netneutrality/http/server/WebApi.java b/src/main/java/netneutrality/http/server/WebApi.java index 002cdc1..dd1f128 100644 --- a/src/main/java/netneutrality/http/server/WebApi.java +++ b/src/main/java/netneutrality/http/server/WebApi.java @@ -6,13 +6,14 @@ package netneutrality.http.server; * This file is closed-source. It should not be distributed or copied in any way or form */ +import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpServer; -import netneutrality.ThreadManager; import netneutrality.thread.BaseThread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.io.OutputStream; import java.net.InetAddress; import java.net.InetSocketAddress; import java.util.concurrent.ExecutorService; @@ -44,6 +45,13 @@ public class WebApi { } + public static void sendStringResponse(HttpExchange exchange, String str) throws IOException { + exchange.sendResponseHeaders(200, str.length()); + OutputStream os = exchange.getResponseBody(); + os.write(str.getBytes()); + os.close(); + } + } diff --git a/src/main/java/netneutrality/music/MusicManager.java b/src/main/java/netneutrality/music/MusicManager.java index a497fb9..4e70f3f 100644 --- a/src/main/java/netneutrality/music/MusicManager.java +++ b/src/main/java/netneutrality/music/MusicManager.java @@ -26,6 +26,7 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class MusicManager { @@ -35,12 +36,11 @@ public class MusicManager { private final AudioPlayer player; private final TrackScheduler scheduler; private final static Map managers = new ConcurrentHashMap<>(); - private final long guildId; - private String topic; + private final String guildId; private MusicManager(Guild guild) { - this.guildId = guild.getIdLong(); + this.guildId = guild.getId(); player = playerManager.createPlayer(); guild.getAudioManager().setSendingHandler(new AudioPlayerSendHandler(player)); scheduler = new TrackScheduler(player); @@ -70,7 +70,7 @@ public class MusicManager { Guild guild = MusicBot.getJDA().getGuildById(guildId); guild.getAudioManager().openAudioConnection(vChannel); - ThreadManager.scheduler.scheduleWithFixedDelay(() -> { + ((ScheduledExecutorService) ThreadManager.getExecutor("music", ThreadManager.ExecutorType.SCHEDULED, 5)).scheduleWithFixedDelay(() -> { if ((getCurrentlyPlaying() == null && scheduler.queue.isEmpty()) || vChannel.getMembers().size() == 0) leave(); }, 120, 120, TimeUnit.SECONDS); diff --git a/src/main/java/netneutrality/utils/ChannelFunction.java b/src/main/java/netneutrality/utils/ChannelFunction.java index e8300d0..27b7e7f 100644 --- a/src/main/java/netneutrality/utils/ChannelFunction.java +++ b/src/main/java/netneutrality/utils/ChannelFunction.java @@ -17,7 +17,7 @@ public class ChannelFunction { if (toDelete >= 2500) return; - ThreadManager.executor.submit(() -> { + ThreadManager.getExecutor("bulk-delete", ThreadManager.ExecutorType.FIXED, 5).submit(() -> { int toDeleteFinal = toDelete + 1; int iteration = toDeleteFinal / 100;