--- jsr166/src/test/tck/ScheduledExecutorTest.java 2017/03/26 02:00:39 1.88 +++ jsr166/src/test/tck/ScheduledExecutorTest.java 2018/01/08 02:04:15 1.96 @@ -11,6 +11,8 @@ import static java.util.concurrent.TimeU import static java.util.concurrent.TimeUnit.SECONDS; import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.concurrent.BlockingQueue; @@ -29,6 +31,7 @@ import java.util.concurrent.ThreadPoolEx import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Stream; import junit.framework.Test; import junit.framework.TestSuite; @@ -72,7 +75,7 @@ public class ScheduledExecutorTest exten Future f = p.schedule(task, timeoutMillis(), MILLISECONDS); assertSame(Boolean.TRUE, f.get()); assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); - assertTrue(done.await(0L, MILLISECONDS)); + assertEquals(0L, done.getCount()); } } @@ -221,110 +224,67 @@ public class ScheduledExecutorTest exten } /** - * execute(null) throws NPE + * Submitting null tasks throws NullPointerException */ - public void testExecuteNull() throws InterruptedException { + public void testNullTaskSubmission() { final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); try (PoolCleaner cleaner = cleaner(p)) { - try { - p.execute(null); - shouldThrow(); - } catch (NullPointerException success) {} + assertNullTaskSubmissionThrowsNullPointerException(p); } } /** - * schedule(null) throws NPE + * Submitted tasks are rejected when shutdown */ - public void testScheduleNull() throws InterruptedException { + public void testSubmittedTasksRejectedWhenShutdown() throws InterruptedException { final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); - try (PoolCleaner cleaner = cleaner(p)) { - try { - TrackedCallable callable = null; - Future f = p.schedule(callable, SHORT_DELAY_MS, MILLISECONDS); - shouldThrow(); - } catch (NullPointerException success) {} - } - } + final ThreadLocalRandom rnd = ThreadLocalRandom.current(); + final CountDownLatch threadsStarted = new CountDownLatch(p.getCorePoolSize()); + final CountDownLatch done = new CountDownLatch(1); + final Runnable r = () -> { + threadsStarted.countDown(); + for (;;) { + try { + done.await(); + return; + } catch (InterruptedException shutdownNowDeliberatelyIgnored) {} + }}; + final Callable c = () -> { + threadsStarted.countDown(); + for (;;) { + try { + done.await(); + return Boolean.TRUE; + } catch (InterruptedException shutdownNowDeliberatelyIgnored) {} + }}; - /** - * execute throws RejectedExecutionException if shutdown - */ - public void testSchedule1_RejectedExecutionException() throws InterruptedException { - final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); - try (PoolCleaner cleaner = cleaner(p)) { - try { - p.shutdown(); - p.schedule(new NoOpRunnable(), - MEDIUM_DELAY_MS, MILLISECONDS); - shouldThrow(); - } catch (RejectedExecutionException success) { - } catch (SecurityException ok) {} - } - } + try (PoolCleaner cleaner = cleaner(p, done)) { + for (int i = p.getCorePoolSize(); i--> 0; ) { + switch (rnd.nextInt(4)) { + case 0: p.execute(r); break; + case 1: assertFalse(p.submit(r).isDone()); break; + case 2: assertFalse(p.submit(r, Boolean.TRUE).isDone()); break; + case 3: assertFalse(p.submit(c).isDone()); break; + } + } - /** - * schedule throws RejectedExecutionException if shutdown - */ - public void testSchedule2_RejectedExecutionException() throws InterruptedException { - final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); - try (PoolCleaner cleaner = cleaner(p)) { - try { - p.shutdown(); - p.schedule(new NoOpCallable(), - MEDIUM_DELAY_MS, MILLISECONDS); - shouldThrow(); - } catch (RejectedExecutionException success) { - } catch (SecurityException ok) {} - } - } + // ScheduledThreadPoolExecutor has an unbounded queue, so never saturated. + await(threadsStarted); - /** - * schedule callable throws RejectedExecutionException if shutdown - */ - public void testSchedule3_RejectedExecutionException() throws InterruptedException { - final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); - try (PoolCleaner cleaner = cleaner(p)) { - try { + if (rnd.nextBoolean()) + p.shutdownNow(); + else p.shutdown(); - p.schedule(new NoOpCallable(), - MEDIUM_DELAY_MS, MILLISECONDS); - shouldThrow(); - } catch (RejectedExecutionException success) { - } catch (SecurityException ok) {} - } - } + // Pool is shutdown, but not yet terminated + assertTaskSubmissionsAreRejected(p); + assertFalse(p.isTerminated()); - /** - * scheduleAtFixedRate throws RejectedExecutionException if shutdown - */ - public void testScheduleAtFixedRate1_RejectedExecutionException() throws InterruptedException { - final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); - try (PoolCleaner cleaner = cleaner(p)) { - try { - p.shutdown(); - p.scheduleAtFixedRate(new NoOpRunnable(), - MEDIUM_DELAY_MS, MEDIUM_DELAY_MS, MILLISECONDS); - shouldThrow(); - } catch (RejectedExecutionException success) { - } catch (SecurityException ok) {} - } - } + done.countDown(); // release blocking tasks + assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); - /** - * scheduleWithFixedDelay throws RejectedExecutionException if shutdown - */ - public void testScheduleWithFixedDelay1_RejectedExecutionException() throws InterruptedException { - final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); - try (PoolCleaner cleaner = cleaner(p)) { - try { - p.shutdown(); - p.scheduleWithFixedDelay(new NoOpRunnable(), - MEDIUM_DELAY_MS, MEDIUM_DELAY_MS, MILLISECONDS); - shouldThrow(); - } catch (RejectedExecutionException success) { - } catch (SecurityException ok) {} + assertTaskSubmissionsAreRejected(p); } + assertEquals(p.getCorePoolSize(), p.getCompletedTaskCount()); } /** @@ -744,11 +704,15 @@ public class ScheduledExecutorTest exten * - setExecuteExistingDelayedTasksAfterShutdownPolicy * - setContinueExistingPeriodicTasksAfterShutdownPolicy */ + @SuppressWarnings("FutureReturnValueIgnored") public void testShutdown_cancellation() throws Exception { - final int poolSize = 2; + final int poolSize = 4; final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(poolSize); + final BlockingQueue q = p.getQueue(); final ThreadLocalRandom rnd = ThreadLocalRandom.current(); + final long delay = rnd.nextInt(2); + final int rounds = rnd.nextInt(1, 3); final boolean effectiveDelayedPolicy; final boolean effectivePeriodicPolicy; final boolean effectiveRemovePolicy; @@ -777,76 +741,150 @@ public class ScheduledExecutorTest exten assertEquals(effectiveRemovePolicy, p.getRemoveOnCancelPolicy()); - // Strategy: Wedge the pool with poolSize "blocker" threads + final boolean periodicTasksContinue = effectivePeriodicPolicy && rnd.nextBoolean(); + + // Strategy: Wedge the pool with one wave of "blocker" tasks, + // then add a second wave that waits in the queue until unblocked. final AtomicInteger ran = new AtomicInteger(0); final CountDownLatch poolBlocked = new CountDownLatch(poolSize); final CountDownLatch unblock = new CountDownLatch(1); - final CountDownLatch periodicLatch1 = new CountDownLatch(2); - final CountDownLatch periodicLatch2 = new CountDownLatch(2); - Runnable task = new CheckedRunnable() { public void realRun() - throws InterruptedException { - poolBlocked.countDown(); - await(unblock); - ran.getAndIncrement(); - }}; - List> blockers = new ArrayList<>(); - List> periodics = new ArrayList<>(); - List> delayeds = new ArrayList<>(); - for (int i = 0; i < poolSize; i++) - blockers.add(p.submit(task)); + final RuntimeException exception = new RuntimeException(); + + class Task implements Runnable { + public void run() { + try { + ran.getAndIncrement(); + poolBlocked.countDown(); + await(unblock); + } catch (Throwable fail) { threadUnexpectedException(fail); } + } + } + + class PeriodicTask extends Task { + PeriodicTask(int rounds) { this.rounds = rounds; } + int rounds; + public void run() { + if (--rounds == 0) super.run(); + // throw exception to surely terminate this periodic task, + // but in a separate execution and in a detectable way. + if (rounds == -1) throw exception; + } + } + + Runnable task = new Task(); + + List> immediates = new ArrayList<>(); + List> delayeds = new ArrayList<>(); + List> periodics = new ArrayList<>(); + + immediates.add(p.submit(task)); + delayeds.add(p.schedule(task, delay, MILLISECONDS)); + periodics.add(p.scheduleAtFixedRate( + new PeriodicTask(rounds), delay, 1, MILLISECONDS)); + periodics.add(p.scheduleWithFixedDelay( + new PeriodicTask(rounds), delay, 1, MILLISECONDS)); + await(poolBlocked); + assertEquals(poolSize, ran.get()); + assertEquals(poolSize, p.getActiveCount()); + assertTrue(q.isEmpty()); + + // Add second wave of tasks. + immediates.add(p.submit(task)); + delayeds.add(p.schedule(task, effectiveDelayedPolicy ? delay : LONG_DELAY_MS, MILLISECONDS)); periodics.add(p.scheduleAtFixedRate( - countDowner(periodicLatch1), 1, 1, MILLISECONDS)); + new PeriodicTask(rounds), delay, 1, MILLISECONDS)); periodics.add(p.scheduleWithFixedDelay( - countDowner(periodicLatch2), 1, 1, MILLISECONDS)); - delayeds.add(p.schedule(task, 1, MILLISECONDS)); + new PeriodicTask(rounds), delay, 1, MILLISECONDS)); + + assertEquals(poolSize, q.size()); + assertEquals(poolSize, ran.get()); + + immediates.forEach( + f -> assertTrue(((ScheduledFuture)f).getDelay(NANOSECONDS) <= 0L)); + + Stream.of(immediates, delayeds, periodics).flatMap(Collection::stream) + .forEach(f -> assertFalse(f.isDone())); - assertTrue(p.getQueue().containsAll(periodics)); - assertTrue(p.getQueue().containsAll(delayeds)); try { p.shutdown(); } catch (SecurityException ok) { return; } assertTrue(p.isShutdown()); + assertTrue(p.isTerminating()); assertFalse(p.isTerminated()); - for (Future periodic : periodics) { - assertTrue(effectivePeriodicPolicy ^ periodic.isCancelled()); - assertTrue(effectivePeriodicPolicy ^ periodic.isDone()); - } - for (Future delayed : delayeds) { - assertTrue(effectiveDelayedPolicy ^ delayed.isCancelled()); - assertTrue(effectiveDelayedPolicy ^ delayed.isDone()); - } - if (testImplementationDetails) { - assertEquals(effectivePeriodicPolicy, - p.getQueue().containsAll(periodics)); - assertEquals(effectiveDelayedPolicy, - p.getQueue().containsAll(delayeds)); - } - unblock.countDown(); // Release all pool threads + if (rnd.nextBoolean()) + assertThrows( + RejectedExecutionException.class, + () -> p.submit(task), + () -> p.schedule(task, 1, SECONDS), + () -> p.scheduleAtFixedRate( + new PeriodicTask(1), 1, 1, SECONDS), + () -> p.scheduleWithFixedDelay( + new PeriodicTask(2), 1, 1, SECONDS)); + + assertTrue(q.contains(immediates.get(1))); + assertTrue(!effectiveDelayedPolicy + ^ q.contains(delayeds.get(1))); + assertTrue(!effectivePeriodicPolicy + ^ q.containsAll(periodics.subList(2, 4))); + + immediates.forEach(f -> assertFalse(f.isDone())); + + assertFalse(delayeds.get(0).isDone()); if (effectiveDelayedPolicy) - for (Future delayed : delayeds) assertNull(delayed.get()); - if (effectivePeriodicPolicy) { - await(periodicLatch1); - await(periodicLatch2); - for (Future periodic : periodics) { - assertTrue(periodic.cancel(false)); - assertTrue(periodic.isCancelled()); - assertTrue(periodic.isDone()); - } + assertFalse(delayeds.get(1).isDone()); + else + assertTrue(delayeds.get(1).isCancelled()); + + if (effectivePeriodicPolicy) + periodics.forEach( + f -> { + assertFalse(f.isDone()); + if (!periodicTasksContinue) { + assertTrue(f.cancel(false)); + assertTrue(f.isCancelled()); + } + }); + else { + periodics.subList(0, 2).forEach(f -> assertFalse(f.isDone())); + periodics.subList(2, 4).forEach(f -> assertTrue(f.isCancelled())); } - for (Future blocker : blockers) assertNull(blocker.get()); + + unblock.countDown(); // Release all pool threads + assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); + assertFalse(p.isTerminating()); assertTrue(p.isTerminated()); - for (Future future : delayeds) { - assertTrue(effectiveDelayedPolicy ^ future.isCancelled()); - assertTrue(future.isDone()); - } - for (Future future : periodics) - assertTrue(future.isCancelled()); - for (Future future : blockers) - assertNull(future.get()); - assertEquals(2 + (effectiveDelayedPolicy ? 1 : 0), ran.get()); + assertTrue(q.isEmpty()); + + Stream.of(immediates, delayeds, periodics).flatMap(Collection::stream) + .forEach(f -> assertTrue(f.isDone())); + + for (Future f : immediates) assertNull(f.get()); + + assertNull(delayeds.get(0).get()); + if (effectiveDelayedPolicy) + assertNull(delayeds.get(1).get()); + else + assertTrue(delayeds.get(1).isCancelled()); + + if (periodicTasksContinue) + periodics.forEach( + f -> { + try { f.get(); } + catch (ExecutionException success) { + assertSame(exception, success.getCause()); + } + catch (Throwable fail) { threadUnexpectedException(fail); } + }); + else + periodics.forEach(f -> assertTrue(f.isCancelled())); + + assertEquals(poolSize + 1 + + (effectiveDelayedPolicy ? 1 : 0) + + (periodicTasksContinue ? 2 : 0), + ran.get()); } /** @@ -886,7 +924,7 @@ public class ScheduledExecutorTest exten } /** - * invokeAny(null) throws NPE + * invokeAny(null) throws NullPointerException */ public void testInvokeAny1() throws Exception { final ExecutorService e = new ScheduledThreadPoolExecutor(2); @@ -899,7 +937,7 @@ public class ScheduledExecutorTest exten } /** - * invokeAny(empty collection) throws IAE + * invokeAny(empty collection) throws IllegalArgumentException */ public void testInvokeAny2() throws Exception { final ExecutorService e = new ScheduledThreadPoolExecutor(2); @@ -912,7 +950,7 @@ public class ScheduledExecutorTest exten } /** - * invokeAny(c) throws NPE if c has null elements + * invokeAny(c) throws NullPointerException if c has null elements */ public void testInvokeAny3() throws Exception { CountDownLatch latch = new CountDownLatch(1); @@ -974,12 +1012,14 @@ public class ScheduledExecutorTest exten } /** - * invokeAll(empty collection) returns empty collection + * invokeAll(empty collection) returns empty list */ public void testInvokeAll2() throws Exception { final ExecutorService e = new ScheduledThreadPoolExecutor(2); + final Collection> emptyCollection + = Collections.emptyList(); try (PoolCleaner cleaner = cleaner(e)) { - List> r = e.invokeAll(new ArrayList>()); + List> r = e.invokeAll(emptyCollection); assertTrue(r.isEmpty()); } } @@ -1042,14 +1082,14 @@ public class ScheduledExecutorTest exten final ExecutorService e = new ScheduledThreadPoolExecutor(2); try (PoolCleaner cleaner = cleaner(e)) { try { - e.invokeAny(null, MEDIUM_DELAY_MS, MILLISECONDS); + e.invokeAny(null, randomTimeout(), randomTimeUnit()); shouldThrow(); } catch (NullPointerException success) {} } } /** - * timed invokeAny(,,null) throws NPE + * timed invokeAny(,,null) throws NullPointerException */ public void testTimedInvokeAnyNullTimeUnit() throws Exception { final ExecutorService e = new ScheduledThreadPoolExecutor(2); @@ -1057,20 +1097,22 @@ public class ScheduledExecutorTest exten List> l = new ArrayList<>(); l.add(new StringTask()); try { - e.invokeAny(l, MEDIUM_DELAY_MS, null); + e.invokeAny(l, randomTimeout(), null); shouldThrow(); } catch (NullPointerException success) {} } } /** - * timed invokeAny(empty collection) throws IAE + * timed invokeAny(empty collection) throws IllegalArgumentException */ public void testTimedInvokeAny2() throws Exception { final ExecutorService e = new ScheduledThreadPoolExecutor(2); + final Collection> emptyCollection + = Collections.emptyList(); try (PoolCleaner cleaner = cleaner(e)) { try { - e.invokeAny(new ArrayList>(), MEDIUM_DELAY_MS, MILLISECONDS); + e.invokeAny(emptyCollection, randomTimeout(), randomTimeUnit()); shouldThrow(); } catch (IllegalArgumentException success) {} } @@ -1087,7 +1129,7 @@ public class ScheduledExecutorTest exten l.add(latchAwaitingStringTask(latch)); l.add(null); try { - e.invokeAny(l, MEDIUM_DELAY_MS, MILLISECONDS); + e.invokeAny(l, randomTimeout(), randomTimeUnit()); shouldThrow(); } catch (NullPointerException success) {} latch.countDown(); @@ -1136,7 +1178,7 @@ public class ScheduledExecutorTest exten final ExecutorService e = new ScheduledThreadPoolExecutor(2); try (PoolCleaner cleaner = cleaner(e)) { try { - e.invokeAll(null, MEDIUM_DELAY_MS, MILLISECONDS); + e.invokeAll(null, randomTimeout(), randomTimeUnit()); shouldThrow(); } catch (NullPointerException success) {} } @@ -1151,20 +1193,22 @@ public class ScheduledExecutorTest exten List> l = new ArrayList<>(); l.add(new StringTask()); try { - e.invokeAll(l, MEDIUM_DELAY_MS, null); + e.invokeAll(l, randomTimeout(), null); shouldThrow(); } catch (NullPointerException success) {} } } /** - * timed invokeAll(empty collection) returns empty collection + * timed invokeAll(empty collection) returns empty list */ public void testTimedInvokeAll2() throws Exception { final ExecutorService e = new ScheduledThreadPoolExecutor(2); + final Collection> emptyCollection + = Collections.emptyList(); try (PoolCleaner cleaner = cleaner(e)) { - List> r = e.invokeAll(new ArrayList>(), - MEDIUM_DELAY_MS, MILLISECONDS); + List> r = + e.invokeAll(emptyCollection, randomTimeout(), randomTimeUnit()); assertTrue(r.isEmpty()); } } @@ -1179,7 +1223,7 @@ public class ScheduledExecutorTest exten l.add(new StringTask()); l.add(null); try { - e.invokeAll(l, MEDIUM_DELAY_MS, MILLISECONDS); + e.invokeAll(l, randomTimeout(), randomTimeUnit()); shouldThrow(); } catch (NullPointerException success) {} } @@ -1265,18 +1309,16 @@ public class ScheduledExecutorTest exten * one-shot task from executing. * https://bugs.openjdk.java.net/browse/JDK-8051859 */ + @SuppressWarnings("FutureReturnValueIgnored") public void testScheduleWithFixedDelay_overflow() throws Exception { final CountDownLatch delayedDone = new CountDownLatch(1); final CountDownLatch immediateDone = new CountDownLatch(1); final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1); try (PoolCleaner cleaner = cleaner(p)) { - final Runnable immediate = new Runnable() { public void run() { - immediateDone.countDown(); - }}; - final Runnable delayed = new Runnable() { public void run() { + final Runnable delayed = () -> { delayedDone.countDown(); - p.submit(immediate); - }}; + p.submit(() -> immediateDone.countDown()); + }; p.scheduleWithFixedDelay(delayed, 0L, Long.MAX_VALUE, SECONDS); await(delayedDone); await(immediateDone);