From 5851f072bfc7ed23bce14613ed847d258161cd2c Mon Sep 17 00:00:00 2001 From: CH4ACKO3 Date: Tue, 6 Oct 2026 18:18:37 +0800 Subject: [PATCH 1/3] Avoid worker notification monitor contention and lost wakeups --- .../impl/task/ParallelTaskExecutor.java | 9 +- .../impl/task/ThreadGroupNotifier.java | 35 ++-- .../impl/task/ThreadGroupNotifierTest.java | 156 ++++++++++++++++++ 3 files changed, 186 insertions(+), 14 deletions(-) create mode 100644 common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java 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); + } } } diff --git a/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java b/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java new file mode 100644 index 0000000000..fdf7459c1e --- /dev/null +++ b/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java @@ -0,0 +1,156 @@ +package dev.engine_room.flywheel.impl.task; + +import static org.junit.jupiter.api.Assertions.*; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(10) +class ThreadGroupNotifierTest { + @Test + void notificationBetweenConditionAndParkIsNotLost() throws Exception { + ThreadGroupNotifier notifier = new ThreadGroupNotifier(); + AtomicBoolean waiting = new AtomicBoolean(true); + CountDownLatch checked = new CountDownLatch(1); + CountDownLatch signalled = new CountDownLatch(1); + Thread worker = new Thread(() -> notifier.awaitNotification(() -> { + boolean result = waiting.get(); + if (result) { + checked.countDown(); + // Do not park here: it would consume the permit under test. + while (signalled.getCount() != 0) Thread.onSpinWait(); + } + return result; + })); + worker.setDaemon(true); + notifier.register(worker); + worker.start(); + try { + assertTrue(checked.await(2, TimeUnit.SECONDS)); + waiting.set(false); + notifier.postNotification(); + signalled.countDown(); + worker.join(2000); + assertFalse(worker.isAlive(), "Notification before park was lost"); + } finally { + waiting.set(false); + signalled.countDown(); + worker.interrupt(); + worker.join(2000); + } + } + + @Test + void senderDoesNotAcquireWorkerMonitor() throws Exception { + ThreadGroupNotifier notifier = new ThreadGroupNotifier(); + CountDownLatch locked = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Thread worker = new Thread(() -> { + synchronized (notifier) { + locked.countDown(); + try { release.await(); } catch (InterruptedException ignored) {} + } + }); + worker.setDaemon(true); + worker.start(); + CountDownLatch sent = new CountDownLatch(1); + Thread sender = new Thread(() -> { notifier.postNotification(); sent.countDown(); }); + sender.setDaemon(true); + try { + assertTrue(locked.await(2, TimeUnit.SECONDS)); + sender.start(); + assertTrue(sent.await(2, TimeUnit.SECONDS)); + } finally { + release.countDown(); + worker.join(2000); + sender.join(2000); + } + } + + @Test + void idleWorkersDoNotPollAndAllWakeOnShutdown() throws Exception { + ThreadGroupNotifier notifier = new ThreadGroupNotifier(); + AtomicBoolean running = new AtomicBoolean(true); + AtomicInteger checks = new AtomicInteger(); + Thread[] workers = new Thread[8]; + for (int i = 0; i < workers.length; i++) { + workers[i] = new Thread(() -> notifier.awaitNotification(() -> { + checks.incrementAndGet(); + return running.get(); + })); + workers[i].setDaemon(true); + notifier.register(workers[i]); + workers[i].start(); + } + try { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); + for (Thread worker : workers) { + while (worker.getState() != Thread.State.WAITING && System.nanoTime() < deadline) Thread.yield(); + assertEquals(Thread.State.WAITING, worker.getState()); + } + int before = checks.get(); + Thread.sleep(100); + // Spurious wakes are legal, but regular 10 ms polling is not. + assertTrue(checks.get() - before < workers.length * 2); + running.set(false); + notifier.postNotification(); + for (Thread worker : workers) { + worker.join(2000); + assertFalse(worker.isAlive()); + } + } finally { + running.set(false); + for (Thread worker : workers) worker.interrupt(); + for (Thread worker : workers) worker.join(2000); + notifier.clear(); + } + } + + @Test + void stoppedConditionDoesNotWaitForAnotherNotification() throws Exception { + ThreadGroupNotifier notifier = new ThreadGroupNotifier(); + Thread worker = new Thread(() -> notifier.awaitNotification(() -> false)); + worker.setDaemon(true); + notifier.register(worker); + worker.start(); + worker.join(2000); + assertFalse(worker.isAlive()); + } + + @Test + void interruptIsConsumedRatherThanSpinning() throws Exception { + ThreadGroupNotifier notifier = new ThreadGroupNotifier(); + AtomicBoolean stillInterrupted = new AtomicBoolean(true); + Thread worker = new Thread(() -> { + Thread.currentThread().interrupt(); + notifier.awaitNotification(() -> true); + stillInterrupted.set(Thread.currentThread().isInterrupted()); + }); + worker.setDaemon(true); + notifier.register(worker); + worker.start(); + worker.join(2000); + assertFalse(worker.isAlive()); + assertFalse(stillInterrupted.get()); + } + + @Test + void executorCanStopAndRestartAfterConcurrentWork() throws Exception { + ParallelTaskExecutor executor = new ParallelTaskExecutor("NotifierTest", 8); + for (int round = 0; round < 5; round++) { + executor.startWorkers(); + try { + CountDownLatch done = new CountDownLatch(10000); + for (int i = 0; i < 10000; i++) executor.execute(done::countDown); + assertTrue(done.await(2, TimeUnit.SECONDS)); + } finally { + executor.stopWorkers(); + } + } + } +} From 6b56aaec0e8faebf6d11d0e757b2702d5c3e3673 Mon Sep 17 00:00:00 2001 From: CH4ACKO3 Date: Sat, 10 Oct 2026 12:16:10 +0800 Subject: [PATCH 2/3] Keep notifier tests focused on regressions --- .../impl/task/ThreadGroupNotifierTest.java | 54 ------------------- 1 file changed, 54 deletions(-) diff --git a/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java b/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java index fdf7459c1e..33ec4e7a3f 100644 --- a/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java +++ b/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java @@ -5,7 +5,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -72,45 +71,6 @@ void senderDoesNotAcquireWorkerMonitor() throws Exception { } } - @Test - void idleWorkersDoNotPollAndAllWakeOnShutdown() throws Exception { - ThreadGroupNotifier notifier = new ThreadGroupNotifier(); - AtomicBoolean running = new AtomicBoolean(true); - AtomicInteger checks = new AtomicInteger(); - Thread[] workers = new Thread[8]; - for (int i = 0; i < workers.length; i++) { - workers[i] = new Thread(() -> notifier.awaitNotification(() -> { - checks.incrementAndGet(); - return running.get(); - })); - workers[i].setDaemon(true); - notifier.register(workers[i]); - workers[i].start(); - } - try { - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); - for (Thread worker : workers) { - while (worker.getState() != Thread.State.WAITING && System.nanoTime() < deadline) Thread.yield(); - assertEquals(Thread.State.WAITING, worker.getState()); - } - int before = checks.get(); - Thread.sleep(100); - // Spurious wakes are legal, but regular 10 ms polling is not. - assertTrue(checks.get() - before < workers.length * 2); - running.set(false); - notifier.postNotification(); - for (Thread worker : workers) { - worker.join(2000); - assertFalse(worker.isAlive()); - } - } finally { - running.set(false); - for (Thread worker : workers) worker.interrupt(); - for (Thread worker : workers) worker.join(2000); - notifier.clear(); - } - } - @Test void stoppedConditionDoesNotWaitForAnotherNotification() throws Exception { ThreadGroupNotifier notifier = new ThreadGroupNotifier(); @@ -139,18 +99,4 @@ void interruptIsConsumedRatherThanSpinning() throws Exception { assertFalse(stillInterrupted.get()); } - @Test - void executorCanStopAndRestartAfterConcurrentWork() throws Exception { - ParallelTaskExecutor executor = new ParallelTaskExecutor("NotifierTest", 8); - for (int round = 0; round < 5; round++) { - executor.startWorkers(); - try { - CountDownLatch done = new CountDownLatch(10000); - for (int i = 0; i < 10000; i++) executor.execute(done::countDown); - assertTrue(done.await(2, TimeUnit.SECONDS)); - } finally { - executor.stopWorkers(); - } - } - } } From 68f7a1804463b4e01b05d726e9cb931eff191519 Mon Sep 17 00:00:00 2001 From: CH4ACKO3 Date: Sat, 10 Oct 2026 18:57:16 +0800 Subject: [PATCH 3/3] Remove tests added for the small fix --- .../impl/task/ThreadGroupNotifierTest.java | 102 ------------------ 1 file changed, 102 deletions(-) delete mode 100644 common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java diff --git a/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java b/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java deleted file mode 100644 index 33ec4e7a3f..0000000000 --- a/common/src/test/java/dev/engine_room/flywheel/impl/task/ThreadGroupNotifierTest.java +++ /dev/null @@ -1,102 +0,0 @@ -package dev.engine_room.flywheel.impl.task; - -import static org.junit.jupiter.api.Assertions.*; - -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; - -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.Timeout; - -@Timeout(10) -class ThreadGroupNotifierTest { - @Test - void notificationBetweenConditionAndParkIsNotLost() throws Exception { - ThreadGroupNotifier notifier = new ThreadGroupNotifier(); - AtomicBoolean waiting = new AtomicBoolean(true); - CountDownLatch checked = new CountDownLatch(1); - CountDownLatch signalled = new CountDownLatch(1); - Thread worker = new Thread(() -> notifier.awaitNotification(() -> { - boolean result = waiting.get(); - if (result) { - checked.countDown(); - // Do not park here: it would consume the permit under test. - while (signalled.getCount() != 0) Thread.onSpinWait(); - } - return result; - })); - worker.setDaemon(true); - notifier.register(worker); - worker.start(); - try { - assertTrue(checked.await(2, TimeUnit.SECONDS)); - waiting.set(false); - notifier.postNotification(); - signalled.countDown(); - worker.join(2000); - assertFalse(worker.isAlive(), "Notification before park was lost"); - } finally { - waiting.set(false); - signalled.countDown(); - worker.interrupt(); - worker.join(2000); - } - } - - @Test - void senderDoesNotAcquireWorkerMonitor() throws Exception { - ThreadGroupNotifier notifier = new ThreadGroupNotifier(); - CountDownLatch locked = new CountDownLatch(1); - CountDownLatch release = new CountDownLatch(1); - Thread worker = new Thread(() -> { - synchronized (notifier) { - locked.countDown(); - try { release.await(); } catch (InterruptedException ignored) {} - } - }); - worker.setDaemon(true); - worker.start(); - CountDownLatch sent = new CountDownLatch(1); - Thread sender = new Thread(() -> { notifier.postNotification(); sent.countDown(); }); - sender.setDaemon(true); - try { - assertTrue(locked.await(2, TimeUnit.SECONDS)); - sender.start(); - assertTrue(sent.await(2, TimeUnit.SECONDS)); - } finally { - release.countDown(); - worker.join(2000); - sender.join(2000); - } - } - - @Test - void stoppedConditionDoesNotWaitForAnotherNotification() throws Exception { - ThreadGroupNotifier notifier = new ThreadGroupNotifier(); - Thread worker = new Thread(() -> notifier.awaitNotification(() -> false)); - worker.setDaemon(true); - notifier.register(worker); - worker.start(); - worker.join(2000); - assertFalse(worker.isAlive()); - } - - @Test - void interruptIsConsumedRatherThanSpinning() throws Exception { - ThreadGroupNotifier notifier = new ThreadGroupNotifier(); - AtomicBoolean stillInterrupted = new AtomicBoolean(true); - Thread worker = new Thread(() -> { - Thread.currentThread().interrupt(); - notifier.awaitNotification(() -> true); - stillInterrupted.set(Thread.currentThread().isInterrupted()); - }); - worker.setDaemon(true); - notifier.register(worker); - worker.start(); - worker.join(2000); - assertFalse(worker.isAlive()); - assertFalse(stillInterrupted.get()); - } - -}