import dev.modzuozhi.core.compat.TaskBarrier; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; public final class TaskBarrierRegression { public static void main(String[] args) throws Exception { // The old one-shot latch is permanently open after the first submission completes. CountDownLatch old = new CountDownLatch(1); old.countDown(); if (!old.await(0, TimeUnit.MILLISECONDS)) throw new AssertionError(); TaskBarrier barrier = new TaskBarrier(); barrier.register(); barrier.complete(); // Fast task finishes while producer is still submitting. barrier.register(); // A later task MUST still be waited for. CountDownLatch warned = new CountDownLatch(1); AtomicBoolean returned = new AtomicBoolean(); AtomicBoolean interruptRestored = new AtomicBoolean(); Thread waiter = new Thread(() -> { barrier.await(30, warned::countDown); returned.set(true); interruptRestored.set(Thread.currentThread().isInterrupted()); }); waiter.setDaemon(true); waiter.start(); waiter.interrupt(); if (!warned.await(2, TimeUnit.SECONDS)) throw new AssertionError("No slow-task diagnostic"); if (returned.get()) throw new AssertionError("Timeout/interruption abandoned a running task"); barrier.complete(); waiter.join(2000); if (waiter.isAlive() || !returned.get() || !interruptRestored.get()) throw new AssertionError("Drain/interrupt restoration failed"); try { barrier.register(); throw new AssertionError("Post-seal submission allowed"); } catch (IllegalStateException expected) {} TaskBarrier empty = new TaskBarrier(); empty.await(30, () -> { throw new AssertionError("Empty barrier waited"); }); try { empty.complete(); throw new AssertionError("Unbalanced completion allowed"); } catch (IllegalStateException expected) {} ExecutorService workers = Executors.newFixedThreadPool(4); try { for (int round = 0; round < 200; round++) { TaskBarrier batch = new TaskBarrier(); int[] writes = new int[100]; for (int i = 0; i < writes.length; i++) { int index = i; batch.register(); workers.execute(() -> { try { writes[index] = 1; } finally { batch.complete(); } }); if ((i & 7) == 0) Thread.yield(); } batch.await(2000, () -> { throw new AssertionError("Stress test stalled"); }); for (int value : writes) if (value != 1) throw new AssertionError("Early return or missing visibility"); } } finally { workers.shutdownNow(); } System.out.println("PASS: submission gaps, timeout, interrupt, empty/sealed barrier, 20,000 concurrent completions"); } }