diff --git a/common/src/main/java/dev/engine_room/flywheel/impl/task/ParallelTaskExecutor.java b/common/src/main/java/dev/engine_room/flywheel/impl/task/ParallelTaskExecutor.java index 066d2ba821..892f1c58ed 100644 --- a/common/src/main/java/dev/engine_room/flywheel/impl/task/ParallelTaskExecutor.java +++ b/common/src/main/java/dev/engine_room/flywheel/impl/task/ParallelTaskExecutor.java @@ -56,6 +56,7 @@ public void startWorkers() { for (int i = 0; i < threadCount; i++) { WorkerThread thread = new WorkerThread(name + " Task Executor #" + i); thread.setPriority(Mth.clamp(Thread.NORM_PRIORITY - 2, Thread.MIN_PRIORITY, Thread.MAX_PRIORITY)); + taskNotifier.register(thread); thread.start(); threads.add(thread); @@ -76,9 +77,7 @@ public void stopWorkers() { FlwImpl.LOGGER.info("Stopping worker threads"); // Notify all worker threads to wake up, where they will then terminate - synchronized (taskNotifier) { - taskNotifier.notifyAll(); - } + taskNotifier.postNotification(); // Wait for every remaining thread to terminate for (Thread thread : threads) { @@ -90,6 +89,7 @@ public void stopWorkers() { } threads.clear(); + taskNotifier.clear(); taskQueue.clear(); waitGroup._reset(); } @@ -190,6 +190,7 @@ private void processTask(Runnable task) { private class WorkerThread extends Thread { private int errorLogLatch = MAX_ERRORS_LOGGED_PER_THREAD; + private final BooleanSupplier shouldWait = () -> running.get() && taskQueue.isEmpty(); public WorkerThread(String name) { super(name); @@ -240,7 +241,7 @@ private void spinThenWait() { Thread.onSpinWait(); } - taskNotifier.awaitNotification(); + taskNotifier.awaitNotification(shouldWait); } } } diff --git a/common/src/main/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifier.java b/common/src/main/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifier.java index 32401b0edd..e1fc77c518 100644 --- a/common/src/main/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifier.java +++ b/common/src/main/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifier.java @@ -1,18 +1,33 @@ package dev.engine_room.flywheel.impl.task; -/** - * Thin wrapper around Java's built-in object synchronization primitives. - */ +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.locks.LockSupport; +import java.util.function.BooleanSupplier; + +/** Signals registered workers without acquiring a monitor held by a worker. */ public class ThreadGroupNotifier { - public synchronized void awaitNotification() { - try { - this.wait(); - } catch (InterruptedException e) { - // we don't care if we're interrupted, just continue. + private final CopyOnWriteArrayList threads = new CopyOnWriteArrayList<>(); + + /** Register before starting the worker; unregister only after it has joined. */ + public void register(Thread thread) { + threads.add(thread); + } + + public void clear() { + threads.clear(); + } + + public void awaitNotification(BooleanSupplier shouldWait) { + while (!Thread.interrupted() && shouldWait.getAsBoolean()) { + // unpark retains a permit if a notification arrives between the condition + // check and park. Recheck after spurious wakeups; no periodic polling. + LockSupport.park(this); } } - public synchronized void postNotification() { - this.notifyAll(); + public void postNotification() { + for (Thread thread : threads) { + LockSupport.unpark(thread); + } } }