Refactor thread management for improved safety and performance
Replaced unbounded thread pools in `ThreadManager` with bounded `ThreadPoolExecutor` to prevent memory exhaustion under load. Refactored shutdown logic to handle exceptions per-executor, set threads to daemon mode in `BaseThreadFactory`, and aligned synchronization patterns across multiple classes (`TaskManagerImpl`, `ChannelMessagePurger`, and `MusicManager`). Addressed lock cleanup races, double unlock issues, and static collection thread-safety concerns.
This commit is contained in:
1 parent
64f4890931
commit
dd1c4dd42f
6 files changed
+136
-30
No files matched your search
@@ -89,13 +89,18 @@ public class TaskManagerImpl implements TaskManager {
|
||||
return tasks.get(id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether a task with the given ID exists. Note: this check spans two collections
|
||||
* ({@code tasks} and {@code pendingTasks}) without a single lock, so the result is
|
||||
* best-effort and may be stale by the time the caller acts on it.
|
||||
*/
|
||||
@Override
|
||||
public boolean hasTask(String id) {
|
||||
return tasks.containsKey(id) || pendingTasks.stream().anyMatch(task -> task.getId().equals(id));
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized Collection<Task> getTasks() {
|
||||
public Collection<Task> getTasks() {
|
||||
return tasks.values();
|
||||
}
|
||||
|
||||
@@ -175,13 +180,9 @@ public class TaskManagerImpl implements TaskManager {
|
||||
|
||||
if (task != null) {
|
||||
String lockId = task.getLockId();
|
||||
ReentrantLock lock = tasksLock.computeIfAbsent(lockId, k -> {
|
||||
if (task instanceof PersistableTask || task instanceof BuiltInTask) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return new ReentrantLock();
|
||||
});
|
||||
ReentrantLock lock = (task instanceof PersistableTask || task instanceof BuiltInTask)
|
||||
? null
|
||||
: tasksLock.computeIfAbsent(lockId, k -> new ReentrantLock());
|
||||
|
||||
try {
|
||||
if (lock != null) {
|
||||
@@ -190,7 +191,6 @@ public class TaskManagerImpl implements TaskManager {
|
||||
|
||||
if (!tasks.containsKey(taskId)) {
|
||||
logger.debug("Task {} is no longer available after aquiring lock", taskId);
|
||||
tasksLock.remove(lockId);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -25,6 +25,7 @@ public class BaseThreadFactory implements ThreadFactory {
|
||||
public Thread newThread(@Nonnull Runnable r) {
|
||||
Thread thread = new Thread(r);
|
||||
thread.setName(name + "-" + threadId.getAndIncrement());
|
||||
thread.setDaemon(true);
|
||||
thread.setUncaughtExceptionHandler(new ExecutorException());
|
||||
return thread;
|
||||
}
|
||||
|
||||
@@ -12,7 +12,9 @@ import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
@@ -50,7 +52,13 @@ public class ThreadManager implements ExecutorProvider {
|
||||
|
||||
@Override
|
||||
public ExecutorService getExecutor(String name) {
|
||||
return executors.computeIfAbsent(name, s -> Executors.newCachedThreadPool(new BaseThreadFactory(s)));
|
||||
return executors.computeIfAbsent(name, s -> new ThreadPoolExecutor(
|
||||
0,
|
||||
Runtime.getRuntime().availableProcessors() * 2,
|
||||
60L, TimeUnit.SECONDS,
|
||||
new LinkedBlockingQueue<>(),
|
||||
new BaseThreadFactory(s)
|
||||
));
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -62,21 +70,20 @@ public class ThreadManager implements ExecutorProvider {
|
||||
public void stop() {
|
||||
logger.info("[Executor] Shutting down...");
|
||||
|
||||
try {
|
||||
for (ExecutorService ex : executors.values()) {
|
||||
for (Map.Entry<String, ExecutorService> entry : executors.entrySet()) {
|
||||
String name = entry.getKey();
|
||||
ExecutorService ex = entry.getValue();
|
||||
try {
|
||||
ex.shutdown();
|
||||
ex.awaitTermination(1, TimeUnit.SECONDS);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
logger.info("[Executor] Running tasks interrupted");
|
||||
} finally {
|
||||
executors.forEach((name, ex) -> {
|
||||
if (!ex.isTerminated()) {
|
||||
if (!ex.awaitTermination(1, TimeUnit.SECONDS)) {
|
||||
logger.warn("[Executor: {}] Forcing shutdown of running tasks..", name);
|
||||
ex.shutdownNow();
|
||||
}
|
||||
});
|
||||
|
||||
} catch (InterruptedException e) {
|
||||
logger.info("[Executor: {}] Running tasks interrupted", name);
|
||||
ex.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
logger.info("[Executor] Shutdown finished!");
|
||||
|
||||
@@ -107,11 +107,11 @@ public class MusicManager implements MusicManagerCallback {
|
||||
return guild.getAudioManager().getConnectedChannel() != null;
|
||||
}
|
||||
|
||||
public AudioTrack getCurrentlyPlaying() {
|
||||
public synchronized AudioTrack getCurrentlyPlaying() {
|
||||
return player.getPlayingTrack();
|
||||
}
|
||||
|
||||
public boolean isCurrentlyPlaying() {
|
||||
public synchronized boolean isCurrentlyPlaying() {
|
||||
return player.getPlayingTrack() != null;
|
||||
}
|
||||
|
||||
@@ -135,11 +135,11 @@ public class MusicManager implements MusicManagerCallback {
|
||||
}
|
||||
}
|
||||
|
||||
public void stopTrack() {
|
||||
public synchronized void stopTrack() {
|
||||
player.stopTrack();
|
||||
}
|
||||
|
||||
public CommandExecutionResult resume() {
|
||||
public synchronized CommandExecutionResult resume() {
|
||||
return getQueueManager().resume();
|
||||
}
|
||||
|
||||
|
||||
@@ -15,9 +15,9 @@ import net.dv8tion.jda.api.utils.messages.MessageEditData;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
@@ -37,7 +37,7 @@ public class ChannelMessagePurger {
|
||||
|
||||
|
||||
private static final Logger logger = LogManager.getLogger(ChannelMessagePurger.class);
|
||||
private static final Map<GuildMessageChannel, ChannelMessagePurger> instances = new HashMap<>();
|
||||
private static final Map<GuildMessageChannel, ChannelMessagePurger> instances = new ConcurrentHashMap<>();
|
||||
private final GuildMessageChannel channel;
|
||||
private final Lock lock;
|
||||
|
||||
@@ -82,7 +82,7 @@ public class ChannelMessagePurger {
|
||||
});
|
||||
}
|
||||
|
||||
private synchronized int countAllChannelMessages(GuildMessageChannel channel) {
|
||||
private int countAllChannelMessages(GuildMessageChannel channel) {
|
||||
logger.info("Guild {}: counting all messages from channel {}", channel.getGuild().getId(), channel.getId());
|
||||
MessageHistory history = channel.getHistory();
|
||||
int totalCount = 0;
|
||||
@@ -99,7 +99,7 @@ public class ChannelMessagePurger {
|
||||
return totalCount;
|
||||
}
|
||||
|
||||
private synchronized void safePurge(CommandContext context, int toPurge) {
|
||||
private void safePurge(CommandContext context, int toPurge) {
|
||||
logger.info("Safe purging {} messages from channel {}", toPurge, channel.getId());
|
||||
|
||||
if (toPurge == 1) {
|
||||
@@ -151,6 +151,5 @@ public class ChannelMessagePurger {
|
||||
builder.addContent("\n").addContent("Messages older than two weeks weren't deleted.");
|
||||
}
|
||||
context.edit(MessageEditData.fromCreateData(builder.build()));
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user