stan/jdk-24
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

8288899: java/util/concurrent/ExecutorService/CloseTest.java failed with "InterruptedException: sleep interrupted"

Browse Source 
Reviewed-by: alanb
This commit is contained in:
Doug Lea 2023-10-27 10:08:59 +00:00
parent b9dcd4b741
commit 667cca9d7a
18 changed files with 3903 additions and 2107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/java.base/share/classes/java/util/concurrent/CountedCompleter.java
View File

@ -524,6 +524,10 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
return pending;
}
final void initPending(int count) {
U.putInt(this, PENDING, count);
}
/**
* Sets the pending count to the given value.
*
@ -724,26 +728,27 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
* processed.
*/
public final void helpComplete(int maxTasks) {
ForkJoinPool.WorkQueue q; Thread t; boolean owned;
if (owned = (t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
ForkJoinPool.WorkQueue q; Thread t; boolean internal;
if (internal =
(t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
q = ((ForkJoinWorkerThread)t).workQueue;
else
q = ForkJoinPool.commonQueue();
if (q != null && maxTasks > 0)
q.helpComplete(this, owned, maxTasks);
q.helpComplete(this, internal, maxTasks);
}
// ForkJoinTask overrides
/**
* Supports ForkJoinTask exception propagation.
*/
@Override
final int trySetException(Throwable ex) {
final void onAuxExceptionSet(Throwable ex) {
CountedCompleter<?> a = this, p = a;
do {} while (isExceptionalStatus(a.trySetThrown(ex)) &&
a.onExceptionalCompletion(ex, p) &&
(a = (p = a).completer) != null && a.status >= 0);
return status;
do {} while (a.onExceptionalCompletion(ex, p) &&
(a = (p = a).completer) != null &&
a.trySetThrown(ex));
}
/**

3050
src/java.base/share/classes/java/util/concurrent/ForkJoinPool.java
View File

File diff suppressed because it is too large Load Diff

1182
src/java.base/share/classes/java/util/concurrent/ForkJoinTask.java
View File

File diff suppressed because it is too large Load Diff

5
src/java.base/share/classes/java/util/concurrent/ForkJoinWorkerThread.java
View File

@ -76,9 +76,8 @@ public class ForkJoinWorkerThread extends Thread {
boolean clearThreadLocals) {
super(group, null, pool.nextWorkerThreadName(), 0L, !clearThreadLocals);
UncaughtExceptionHandler handler = (this.pool = pool).ueh;
this.workQueue = new ForkJoinPool.WorkQueue(this, 0);
if (clearThreadLocals)
workQueue.setClearThreadLocals();
this.workQueue = new ForkJoinPool.WorkQueue(this, 0, (int)pool.config,
clearThreadLocals);
super.setDaemon(true);
if (handler != null)
super.setUncaughtExceptionHandler(handler);

4
test/jdk/ProblemList.txt
View File

@ -724,8 +724,8 @@ com/sun/jdi/InvokeHangTest.java 8218463 linux-al
# jdk_util
java/util/concurrent/forkjoin/AsyncShutdownNow.java 8286352 linux-all,windows-x64
java/util/concurrent/ExecutorService/CloseTest.java 8288899 macosx-aarch64
java/util/Locale/LocaleProvidersRun.java 8268379 macosx-x64
sun/util/locale/provider/CalendarDataRegression.java 8268379 macosx-x64
############################################################################

171
test/jdk/java/util/concurrent/ExecutorService/CloseTest.java
View File

@ -23,9 +23,9 @@
/*
* @test
* @summary Test ExecutorService.close, including default implementation
* @summary Test implementations of ExecutorService.close
* @library ../lib
* @run testng CloseTest
* @run junit CloseTest
*/
import java.time.Duration;
@ -37,37 +37,32 @@ import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.*;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import static org.junit.jupiter.api.Assertions.*;
public class CloseTest {
class CloseTest {
@DataProvider(name = "executors")
public Object[][] executors() {
var defaultThreadFactory = Executors.defaultThreadFactory();
var virtualThreadFactory = Thread.ofVirtual().factory();
return new Object[][] {
// ensures that default close method is tested
{ new DelegatingExecutorService(Executors.newCachedThreadPool()), },
static Stream<ExecutorService> executors() {
return Stream.of(
// ensures that default close method is tested
new DelegatingExecutorService(Executors.newCachedThreadPool()),
// implementations that may override close
{ new ForkJoinPool(), },
{ Executors.newFixedThreadPool(1), },
{ Executors.newCachedThreadPool(), },
{ Executors.newThreadPerTaskExecutor(defaultThreadFactory), },
{ Executors.newThreadPerTaskExecutor(virtualThreadFactory), },
};
// implementations that may override close
Executors.newCachedThreadPool(),
Executors.newVirtualThreadPerTaskExecutor(),
new ForkJoinPool()
);
}
/**
* Test close with no tasks running.
*/
@Test(dataProvider = "executors")
public void testCloseWithNoTasks(ExecutorService executor) throws Exception {
@ParameterizedTest
@MethodSource("executors")
void testCloseWithNoTasks(ExecutorService executor) throws Exception {
executor.close();
assertTrue(executor.isShutdown());
assertTrue(executor.isTerminated());
@ -77,24 +72,109 @@ public class CloseTest {
/**
* Test close with tasks running.
*/
@Test(dataProvider = "executors")
public void testCloseWithRunningTasks(ExecutorService executor) throws Exception {
@ParameterizedTest
@MethodSource("executors")
void testCloseWithRunningTasks(ExecutorService executor) throws Exception {
Phaser phaser = new Phaser(2);
Future<?> future = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
Thread.sleep(Duration.ofMillis(100));
return "foo";
});
phaser.arriveAndAwaitAdvance(); // wait for task to start
executor.close(); // waits for task to complete
assertFalse(Thread.interrupted());
assertTrue(executor.isShutdown());
assertTrue(executor.isTerminated());
assertTrue(executor.awaitTermination(10, TimeUnit.MILLISECONDS));
assertEquals(future.resultNow(), "foo");
assertEquals("foo", future.resultNow());
}
/**
* Test shutdown with tasks running.
*/
@ParameterizedTest
@MethodSource("executors")
void testShutdownWithRunningTasks(ExecutorService executor) throws Exception {
Phaser phaser = new Phaser(2);
Future<?> future = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
Thread.sleep(Duration.ofMillis(100));
return "foo";
});
phaser.arriveAndAwaitAdvance(); // wait for task to start
executor.shutdown();
assertFalse(Thread.interrupted());
assertTrue(executor.isShutdown());
assertTrue(executor.awaitTermination(1, TimeUnit.MINUTES));
assertTrue(executor.isTerminated());
assertEquals("foo", future.resultNow());
}
/**
* Test close with multiple tasks running
*/
@ParameterizedTest
@MethodSource("executors")
void testCloseWith2RunningTasks(ExecutorService executor) throws Exception {
Phaser phaser = new Phaser(3);
Future<?> f1 = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
Thread.sleep(Duration.ofMillis(100));
return "foo";
});
Future<?> f2 = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
Thread.sleep(Duration.ofMillis(100));
return "bar";
});
phaser.arriveAndAwaitAdvance(); // wait for tasks to start
executor.close(); // waits for task to complete
assertFalse(Thread.interrupted());
assertTrue(executor.isShutdown());
assertTrue(executor.isTerminated());
assertTrue(executor.awaitTermination(10, TimeUnit.MILLISECONDS));
assertEquals("foo", f1.resultNow());
assertEquals("bar", f2.resultNow());
}
/**
* Test shutdown with multiple tasks running
*/
@ParameterizedTest
@MethodSource("executors")
void testShutdownWith2RunningTasks(ExecutorService executor) throws Exception {
Phaser phaser = new Phaser(3);
Future<?> f1 = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
Thread.sleep(Duration.ofMillis(100));
return "foo";
});
Future<?> f2 = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
Thread.sleep(Duration.ofMillis(100));
return "bar";
});
phaser.arriveAndAwaitAdvance(); // wait for tasks to start
executor.shutdown();
assertFalse(Thread.interrupted());
assertTrue(executor.isShutdown());
assertTrue(executor.awaitTermination(1, TimeUnit.MINUTES));
assertTrue(executor.isTerminated());
assertEquals("foo", f1.resultNow());
assertEquals("bar", f2.resultNow());
}
/**
* Test close when executor is shutdown but not terminated.
*/
@Test(dataProvider = "executors")
public void testShutdownBeforeClose(ExecutorService executor) throws Exception {
@ParameterizedTest
@MethodSource("executors")
void testShutdownBeforeClose(ExecutorService executor) throws Exception {
Phaser phaser = new Phaser(2);
Future<?> future = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
@ -104,22 +184,22 @@ public class CloseTest {
phaser.arriveAndAwaitAdvance(); // wait for task to start
executor.shutdown(); // shutdown, will not immediately terminate
executor.close();
assertTrue(executor.isShutdown());
assertTrue(executor.isTerminated());
assertTrue(executor.awaitTermination(10, TimeUnit.MILLISECONDS));
assertEquals(future.resultNow(), "foo");
Object s = future.resultNow();
assertEquals("foo", s);
}
/**
* Test close when terminated.
*/
@Test(dataProvider = "executors")
public void testTerminateBeforeClose(ExecutorService executor) throws Exception {
@ParameterizedTest
@MethodSource("executors")
void testTerminateBeforeClose(ExecutorService executor) throws Exception {
executor.shutdown();
assertTrue(executor.isTerminated());
executor.close();
assertTrue(executor.isShutdown());
assertTrue(executor.isTerminated());
@ -129,8 +209,9 @@ public class CloseTest {
/**
* Test invoking close with interrupt status set.
*/
@Test(dataProvider = "executors")
public void testInterruptBeforeClose(ExecutorService executor) throws Exception {
@ParameterizedTest
@MethodSource("executors")
void testInterruptBeforeClose(ExecutorService executor) throws Exception {
Phaser phaser = new Phaser(2);
Future<?> future = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
@ -149,21 +230,29 @@ public class CloseTest {
assertTrue(executor.isShutdown());
assertTrue(executor.isTerminated());
assertTrue(executor.awaitTermination(10, TimeUnit.MILLISECONDS));
expectThrows(ExecutionException.class, future::get);
assertThrows(ExecutionException.class, future::get);
}
/**
* Test interrupting thread blocked in close.
*/
@Test(dataProvider = "executors")
public void testInterruptDuringClose(ExecutorService executor) throws Exception {
@ParameterizedTest
@MethodSource("executors")
void testInterruptDuringClose(ExecutorService executor) throws Exception {
Phaser phaser = new Phaser(2);
Future<?> future = executor.submit(() -> {
phaser.arriveAndAwaitAdvance();
Thread.sleep(Duration.ofDays(1));
return null;
});
phaser.arriveAndAwaitAdvance(); // wait for task to start
// schedule main thread to be interrupted
Thread thread = Thread.currentThread();
new Thread(() -> {
try { Thread.sleep( Duration.ofMillis(500)); } catch (Exception ignore) { }
try {
Thread.sleep( Duration.ofMillis(100));
} catch (Exception ignore) { }
thread.interrupt();
}).start();
try {
@ -175,6 +264,6 @@ public class CloseTest {
assertTrue(executor.isShutdown());
assertTrue(executor.isTerminated());
assertTrue(executor.awaitTermination(10, TimeUnit.MILLISECONDS));
expectThrows(ExecutionException.class, future::get);
assertThrows(ExecutionException.class, future::get);
}
}

787
test/jdk/java/util/concurrent/ExecutorService/InvokeTest.java Normal file
View File

@ -0,0 +1,787 @@
/*
* Copyright (c) 2023, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
/**
* @test
* @summary Test implementations of ExecutorService.invokeAll/invokeAny
* @run junit InvokeTest
*/
import java.time.Duration;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
import static java.lang.Thread.State.*;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import static org.junit.jupiter.api.Assertions.*;
class InvokeTest {
private static ScheduledExecutorService scheduler;
@BeforeAll
static void setup() throws Exception {
scheduler = Executors.newSingleThreadScheduledExecutor();
}
@AfterAll
static void shutdown() {
scheduler.shutdown();
}
private static Stream<ExecutorService> executors() {
return Stream.of(
Executors.newCachedThreadPool(),
Executors.newVirtualThreadPerTaskExecutor(),
new ForkJoinPool()
);
}
/**
* Test invokeAny where all tasks complete normally.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAny1(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> "bar";
String result = executor.invokeAny(List.of(task1, task2));
assertTrue(Set.of("foo", "bar").contains(result));
}
}
/**
* Test invokeAny where all tasks complete normally. The completion of the
* first task should cancel remaining tasks.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAny2(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
var task2Started = new AtomicBoolean();
var task2Interrupted = new CountDownLatch(1);
Callable<String> task2 = () -> {
task2Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task2Interrupted.countDown();
}
return null;
};
String result = executor.invokeAny(List.of(task1, task2));
assertEquals("foo", result);
// if task2 started then it should have been interrupted
if (task2Started.get()) {
task2Interrupted.await();
}
}
}
/**
* Test invokeAny where all tasks complete with exception.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAny3(ExecutorService executor) throws Exception {
try (executor) {
class FooException extends Exception { }
Callable<String> task1 = () -> { throw new FooException(); };
Callable<String> task2 = () -> { throw new FooException(); };
try {
executor.invokeAny(List.of(task1, task2));
fail("invokeAny did not throw");
} catch (ExecutionException e) {
Throwable cause = e.getCause();
assertTrue(cause instanceof FooException);
}
}
}
/**
* Test invokeAny where all tasks complete with exception. The completion
* of the last task is delayed.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAny4(ExecutorService executor) throws Exception {
try (executor) {
class FooException extends Exception { }
Callable<String> task1 = () -> { throw new FooException(); };
Callable<String> task2 = () -> {
Thread.sleep(Duration.ofMillis(50));
throw new FooException();
};
try {
executor.invokeAny(List.of(task1, task2));
fail("invokeAny did not throw");
} catch (ExecutionException e) {
Throwable cause = e.getCause();
assertTrue(cause instanceof FooException);
}
}
}
/**
* Test invokeAny where some, not all, tasks complete normally.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAny5(ExecutorService executor) throws Exception {
try (executor) {
class FooException extends Exception { }
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> { throw new FooException(); };
String result = executor.invokeAny(List.of(task1, task2));
assertEquals("foo", result);
}
}
/**
* Test invokeAny where some, not all, tasks complete normally. The first
* task to complete normally is delayed.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAny6(ExecutorService executor) throws Exception {
try (executor) {
class FooException extends Exception { }
Callable<String> task1 = () -> {
Thread.sleep(Duration.ofMillis(50));
return "foo";
};
Callable<String> task2 = () -> { throw new FooException(); };
String result = executor.invokeAny(List.of(task1, task2));
assertEquals("foo", result);
}
}
/**
* Test timed-invokeAny where all tasks complete normally before the timeout.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyWithTimeout1(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> "bar";
String result = executor.invokeAny(List.of(task1, task2), 1, TimeUnit.MINUTES);
assertTrue(Set.of("foo", "bar").contains(result));
}
}
/**
* Test timed-invokeAny where one task completes normally before the timeout.
* The remaining tests should be cancelled.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyWithTimeout2(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
var task2Started = new AtomicBoolean();
var task2Interrupted = new CountDownLatch(1);
Callable<String> task2 = () -> {
task2Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task2Interrupted.countDown();
}
return null;
};
String result = executor.invokeAny(List.of(task1, task2), 1, TimeUnit.MINUTES);
assertEquals("foo", result);
// if task2 started then it should have been interrupted
if (task2Started.get()) {
task2Interrupted.await();
}
}
}
/**
* Test timed-invokeAny where timeout expires before any task completes.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyWithTimeout3(ExecutorService executor) throws Exception {
try (executor) {
var task1Started = new AtomicBoolean();
var task1Interrupted = new CountDownLatch(1);
Callable<String> task1 = () -> {
task1Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task1Interrupted.countDown();
}
return null;
};
var task2Started = new AtomicBoolean();
var task2Interrupted = new CountDownLatch(1);
Callable<String> task2 = () -> {
task2Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task2Interrupted.countDown();
}
return null;
};
// invokeAny should throw TimeoutException
assertThrows(TimeoutException.class,
() -> executor.invokeAny(List.of(task1, task2), 100, TimeUnit.MILLISECONDS));
// tasks that started should be interrupted
if (task1Started.get()) {
task1Interrupted.await();
}
if (task2Started.get()) {
task2Interrupted.await();
}
}
}
/**
* Test invokeAny with interrupt status set.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyWithInterruptSet(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> {
Thread.sleep(Duration.ofMinutes(1));
return "foo";
};
Callable<String> task2 = () -> {
Thread.sleep(Duration.ofMinutes(1));
return "bar";
};
Thread.currentThread().interrupt();
try {
executor.invokeAny(List.of(task1, task2));
fail("invokeAny did not throw");
} catch (InterruptedException expected) {
assertFalse(Thread.currentThread().isInterrupted());
} finally {
Thread.interrupted(); // clear interrupt
}
}
}
/**
* Test interrupting a thread blocked in invokeAny.
*/
@ParameterizedTest
@MethodSource("executors")
void testInterruptInvokeAny(ExecutorService executor) throws Exception {
try (executor) {
var task1Started = new AtomicBoolean();
var task1Interrupted = new CountDownLatch(1);
Callable<String> task1 = () -> {
task1Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task1Interrupted.countDown();
}
return null;
};
var task2Started = new AtomicBoolean();
var task2Interrupted = new CountDownLatch(1);
Callable<String> task2 = () -> {
task2Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task2Interrupted.countDown();
}
return null;
};
scheduleInterruptAt("invokeAny");
try {
executor.invokeAny(List.of(task1, task2));
fail("invokeAny did not throw");
} catch (InterruptedException expected) {
assertFalse(Thread.currentThread().isInterrupted());
} finally {
Thread.interrupted(); // clear interrupt
}
// tasks that started should be interrupted
if (task1Started.get()) {
task1Interrupted.await();
}
if (task2Started.get()) {
task2Interrupted.await();
}
}
}
/**
* Test invokeAny after ExecutorService has been shutdown.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyAfterShutdown(ExecutorService executor) throws Exception {
executor.shutdown();
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> "bar";
assertThrows(RejectedExecutionException.class,
() -> executor.invokeAny(List.of(task1, task2)));
}
/**
* Test invokeAny with empty collection.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyEmpty1(ExecutorService executor) throws Exception {
try (executor) {
assertThrows(IllegalArgumentException.class, () -> executor.invokeAny(List.of()));
}
}
/**
* Test timed-invokeAny with empty collection.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyEmpty2(ExecutorService executor) throws Exception {
try (executor) {
assertThrows(IllegalArgumentException.class,
() -> executor.invokeAny(List.of(), 1, TimeUnit.MINUTES));
}
}
/**
* Test invokeAny with null.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyNull1(ExecutorService executor)throws Exception {
try (executor) {
assertThrows(NullPointerException.class, () -> executor.invokeAny(null));
}
}
/**
* Test invokeAny with null element
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAnyNull2(ExecutorService executor)throws Exception {
try (executor) {
List<Callable<String>> list = new ArrayList<>();
list.add(() -> "foo");
list.add(null);
assertThrows(NullPointerException.class, () -> executor.invokeAny(null));
}
}
/**
* Test invokeAll where all tasks complete normally.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAll1(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> {
Thread.sleep(Duration.ofMillis(50));
return "bar";
};
List<Future<String>> futures = executor.invokeAll(List.of(task1, task2));
assertTrue(futures.size() == 2);
// check results
List<String> results = futures.stream().map(Future::resultNow).toList();
assertEquals(results, List.of("foo", "bar"));
}
}
/**
* Test invokeAll where all tasks complete with exception.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAll2(ExecutorService executor) throws Exception {
try (executor) {
class FooException extends Exception { }
class BarException extends Exception { }
Callable<String> task1 = () -> { throw new FooException(); };
Callable<String> task2 = () -> {
Thread.sleep(Duration.ofMillis(50));
throw new BarException();
};
List<Future<String>> futures = executor.invokeAll(List.of(task1, task2));
assertTrue(futures.size() == 2);
// check results
Throwable e1 = assertThrows(ExecutionException.class, () -> futures.get(0).get());
assertTrue(e1.getCause() instanceof FooException);
Throwable e2 = assertThrows(ExecutionException.class, () -> futures.get(1).get());
assertTrue(e2.getCause() instanceof BarException);
}
}
/**
* Test invokeAll where all tasks complete normally before the timeout expires.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAll3(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> {
Thread.sleep(Duration.ofMillis(50));
return "bar";
};
List<Future<String>> futures = executor.invokeAll(List.of(task1, task2), 1, TimeUnit.MINUTES);
assertTrue(futures.size() == 2);
// check results
List<String> results = futures.stream().map(Future::resultNow).toList();
assertEquals(results, List.of("foo", "bar"));
}
}
/**
* Test invokeAll where some tasks do not complete before the timeout expires.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAll4(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
var task2Started = new AtomicBoolean();
var task2Interrupted = new CountDownLatch(1);
Callable<String> task2 = () -> {
task2Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task2Interrupted.countDown();
}
return null;
};
List<Future<String>> futures = executor.invokeAll(List.of(task1, task2), 1, TimeUnit.SECONDS);
assertTrue(futures.size() == 2);
// task1 should be done
assertTrue(futures.get(0).isDone());
// task2 should be cancelled and interrupted
assertTrue(futures.get(1).isCancelled());
// if task2 started then it should have been interrupted
if (task2Started.get()) {
task2Interrupted.await();
}
}
}
/**
* Test invokeAll with interrupt status set.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllInterrupt1(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> {
Thread.sleep(Duration.ofMinutes(1));
return "bar";
};
Thread.currentThread().interrupt();
try {
executor.invokeAll(List.of(task1, task2));
fail("invokeAll did not throw");
} catch (InterruptedException expected) {
assertFalse(Thread.currentThread().isInterrupted());
} finally {
Thread.interrupted(); // clear interrupt
}
}
}
/**
* Test timed-invokeAll with interrupt status set.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllInterrupt3(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> {
Thread.sleep(Duration.ofMinutes(1));
return "bar";
};
Thread.currentThread().interrupt();
try {
executor.invokeAll(List.of(task1, task2), 1, TimeUnit.MINUTES);
fail("invokeAll did not throw");
} catch (InterruptedException expected) {
assertFalse(Thread.currentThread().isInterrupted());
} finally {
Thread.interrupted(); // clear interrupt
}
}
}
/**
* Test interrupt with thread blocked in invokeAll.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllInterrupt4(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
var task2Started = new AtomicBoolean();
var task2Interrupted = new CountDownLatch(1);
Callable<String> task2 = () -> {
task2Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task2Interrupted.countDown();
}
return null;
};
scheduleInterruptAt("invokeAll");
try {
executor.invokeAll(List.of(task1, task2));
fail("invokeAll did not throw");
} catch (InterruptedException expected) {
assertFalse(Thread.currentThread().isInterrupted());
} finally {
Thread.interrupted(); // clear interrupt
}
// if task2 started then it should have been interrupted
if (task2Started.get()) {
task2Interrupted.await();
}
}
}
/**
* Test interrupt with thread blocked in timed-invokeAll.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllInterrupt6(ExecutorService executor) throws Exception {
try (executor) {
Callable<String> task1 = () -> "foo";
var task2Started = new AtomicBoolean();
var task2Interrupted = new CountDownLatch(1);
Callable<String> task2 = () -> {
task2Started.set(true);
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
task2Interrupted.countDown();
}
return null;
};
scheduleInterruptAt("invokeAll");
try {
executor.invokeAll(List.of(task1, task2), 1, TimeUnit.MINUTES);
fail("invokeAll did not throw");
} catch (InterruptedException expected) {
assertFalse(Thread.currentThread().isInterrupted());
} finally {
Thread.interrupted(); // clear interrupt
}
// if task2 started then it should have been interrupted
if (task2Started.get()) {
task2Interrupted.await();
}
}
}
/**
* Test invokeAll after ExecutorService has been shutdown.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllAfterShutdown1(ExecutorService executor) throws Exception {
executor.shutdown();
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> "bar";
assertThrows(RejectedExecutionException.class,
() -> executor.invokeAll(List.of(task1, task2)));
}
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllAfterShutdown2(ExecutorService executor) throws Exception {
executor.shutdown();
Callable<String> task1 = () -> "foo";
Callable<String> task2 = () -> "bar";
assertThrows(RejectedExecutionException.class,
() -> executor.invokeAll(List.of(task1, task2), 1, TimeUnit.SECONDS));
}
/**
* Test invokeAll with empty collection.
*/
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllEmpty1(ExecutorService executor) throws Exception {
try (executor) {
List<Future<Object>> list = executor.invokeAll(List.of());
assertTrue(list.size() == 0);
}
}
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllEmpty2(ExecutorService executor) throws Exception {
try (executor) {
List<Future<Object>> list = executor.invokeAll(List.of(), 1, TimeUnit.SECONDS);
assertTrue(list.size() == 0);
}
}
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllNull1(ExecutorService executor)throws Exception {
try (executor) {
assertThrows(NullPointerException.class, () -> executor.invokeAll(null));
}
}
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllNull2(ExecutorService executor)throws Exception {
try (executor) {
List<Callable<String>> tasks = new ArrayList<>();
tasks.add(() -> "foo");
tasks.add(null);
assertThrows(NullPointerException.class, () -> executor.invokeAll(tasks));
}
}
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllNull3(ExecutorService executor)throws Exception {
try (executor) {
assertThrows(NullPointerException.class,
() -> executor.invokeAll(null, 1, TimeUnit.SECONDS));
}
}
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllNull4(ExecutorService executor)throws Exception {
try (executor) {
Callable<String> task = () -> "foo";
assertThrows(NullPointerException.class,
() -> executor.invokeAll(List.of(task), 1, null));
}
}
@ParameterizedTest
@MethodSource("executors")
void testInvokeAllNull5(ExecutorService executor)throws Exception {
try (executor) {
List<Callable<String>> tasks = new ArrayList<>();
tasks.add(() -> "foo");
tasks.add(null);
assertThrows(NullPointerException.class,
() -> executor.invokeAll(tasks, 1, TimeUnit.SECONDS));
}
}
/**
* Schedules the current thread to be interrupted when it waits (timed or untimed)
* at the given method name.
*/
private void scheduleInterruptAt(String methodName) {
Thread target = Thread.currentThread();
scheduler.submit(() -> {
try {
boolean found = false;
while (!found) {
Thread.State state = target.getState();
assertTrue(state != TERMINATED);
if ((state == WAITING || state == TIMED_WAITING)
&& contains(target.getStackTrace(), methodName)) {
found = true;
} else {
Thread.sleep(20);
}
}
target.interrupt();
} catch (Exception e) {
e.printStackTrace();
}
});
}
/**
* Returns true if the given stack trace contains an element for the given method name.
*/
private boolean contains(StackTraceElement[] stack, String methodName) {
return Arrays.stream(stack)
.anyMatch(e -> methodName.equals(e.getMethodName()));
}
}

370
test/jdk/java/util/concurrent/ExecutorService/SubmitTest.java Normal file
View File

@ -0,0 +1,370 @@
/*
* Copyright (c) 2023, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
/**
* @test
* @summary Test implementations of ExecutorService.submit/execute
* @run junit SubmitTest
*/
import java.time.Duration;
import java.util.concurrent.*;
import java.util.stream.Stream;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import static org.junit.jupiter.api.Assertions.*;
class SubmitTest {
private static Stream<ExecutorService> executors() {
return Stream.of(
Executors.newCachedThreadPool(),
Executors.newVirtualThreadPerTaskExecutor(),
new ForkJoinPool()
);
}
/**
* Test submit(Runnable) executes the task.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitRunnable(ExecutorService executor) throws Exception {
try (executor) {
var latch = new CountDownLatch(1);
Future<?> future = executor.submit(latch::countDown);
latch.await();
assertNull(future.get());
}
}
/**
* Test submit(Runnable) throws if executor is shutdown.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitRunnableAfterShutdown(ExecutorService executor) {
executor.shutdown();
assertThrows(RejectedExecutionException.class, () -> executor.submit(() -> { }));
}
/**
* Test task submitted with submit(Runnable) is not interrupted by cancel(false).
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitRunnableWithCancelFalse(ExecutorService executor) throws Exception {
try (executor) {
var started = new CountDownLatch(1);
var stop = new CountDownLatch(1);
var done = new CountDownLatch(1);
Future<?> future = executor.submit(() -> {
started.countDown();
try {
stop.await();
} catch (InterruptedException e) {
// ignore
} finally {
done.countDown();
}
});
// wait for task to start
started.await();
// cancel(false), task should not be interrupted
future.cancel(false);
assertFalse(done.await(500, TimeUnit.MILLISECONDS));
// let task finish
stop.countDown();
}
}
/**
* Test task submitted with submit(Runnable) is interrupted by cancel(true).
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitRunnableWithCancelTrue(ExecutorService executor) throws Exception {
try (executor) {
var started = new CountDownLatch(1);
var interrupted = new CountDownLatch(1);
Future<?> future = executor.submit(() -> {
started.countDown();
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
interrupted.countDown();
}
});
// wait for task to start
started.await();
// cancel(true), task should be interrupted
future.cancel(true);
interrupted.await();
}
}
/**
* Test task submitted with submit(Runnable) is interrupted if executor is
* stopped with shutdownNow.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitRunnableWithShutdownNow(ExecutorService executor) throws Exception {
try (executor) {
var started = new CountDownLatch(1);
var interrupted = new CountDownLatch(1);
Future<?> future = executor.submit(() -> {
started.countDown();
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
interrupted.countDown();
}
});
// wait for task to start
started.await();
// shutdown forcefully, task should be interrupted
executor.shutdownNow();
interrupted.await();
}
}
/**
* Test submit(Runnable) throws if task is null.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitRunnableNull(ExecutorService executor) {
try (executor) {
Runnable nullTask = null;
assertThrows(NullPointerException.class, () -> executor.submit(nullTask));
assertThrows(NullPointerException.class, () -> executor.submit(nullTask, Void.class));
}
}
//
/**
* Test submit(Callable) executes the task.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitCallable(ExecutorService executor) throws Exception {
try (executor) {
var latch = new CountDownLatch(1);
Future<String> future = executor.submit(() -> {
latch.countDown();
return "foo";
});
latch.await();
assertEquals("foo", future.get());
}
}
/**
* Test submit(Callable) throws if executor is shutdown.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitCallableAfterShutdown(ExecutorService executor) {
executor.shutdown();
assertThrows(RejectedExecutionException.class, () -> executor.submit(() -> null));
}
/**
* Test task submitted with submit(Callable) is not interrupted by cancel(false).
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitCallableWithCancelFalse(ExecutorService executor) throws Exception {
try (executor) {
var started = new CountDownLatch(1);
var stop = new CountDownLatch(1);
var done = new CountDownLatch(1);
Future<Void> future = executor.submit(() -> {
started.countDown();
try {
stop.await();
} finally {
done.countDown();
}
return null;
});
// wait for task to start
started.await();
// cancel(false), task should not be interrupted
future.cancel(false);
assertFalse(done.await(500, TimeUnit.MILLISECONDS));
// let task finish
stop.countDown();
}
}
/**
* Test task submitted with submit(Callable) is interrupted by cancel(true).
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitCallableWithCancelTrue(ExecutorService executor) throws Exception {
try (executor) {
var started = new CountDownLatch(1);
var interrupted = new CountDownLatch(1);
Future<Void> future = executor.submit(() -> {
started.countDown();
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
interrupted.countDown();
}
return null;
});
// wait for task to start
started.await();
// cancel(true), task should be interrupted
future.cancel(true);
interrupted.await();
}
}
/**
* Test task submitted with submit(Callable) is interrupted if executor is
* stopped with shutdownNow.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitCallableWithShutdownNow(ExecutorService executor) throws Exception {
try (executor) {
var started = new CountDownLatch(1);
var interrupted = new CountDownLatch(1);
Future<Void> future = executor.submit(() -> {
started.countDown();
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
interrupted.countDown();
}
return null;
});
// wait for task to start
started.await();
// shutdown forcefully, task should be interrupted
executor.shutdownNow();
interrupted.await();
}
}
/**
* Test submit(Callable) throws if task is null.
*/
@ParameterizedTest
@MethodSource("executors")
void testSubmitCallableNull(ExecutorService executor) {
try (executor) {
Callable<Void> nullTask = null;
assertThrows(NullPointerException.class, () -> executor.submit(nullTask));
}
}
//
/**
* Test execute(Runnable) executes the task.
*/
@ParameterizedTest
@MethodSource("executors")
void testExecute(ExecutorService executor) throws Exception {
try (executor) {
var latch = new CountDownLatch(1);
executor.execute(latch::countDown);
latch.await();
}
}
/**
* Test execute(Runnable) throws if executor is shutdown.
*/
@ParameterizedTest
@MethodSource("executors")
void testExecuteAfterShutdown(ExecutorService executor) {
executor.shutdown();
assertThrows(RejectedExecutionException.class, () -> executor.execute(() -> { }));
}
/**
* Test task submitted with execute(Runnable) is interrupted if executor is
* stopped with shutdownNow.
*/
@ParameterizedTest
@MethodSource("executors")
void testExecuteWithShutdownNow(ExecutorService executor) throws Exception {
try (executor) {
var started = new CountDownLatch(1);
var interrupted = new CountDownLatch(1);
executor.execute(() -> {
started.countDown();
try {
Thread.sleep(Duration.ofDays(1));
} catch (InterruptedException e) {
interrupted.countDown();
}
});
// wait for task to start
started.await();
// shutdown forcefully, task should be interrupted
executor.shutdownNow();
interrupted.await();
}
}
/**
* Test execute(Runnable) throws if task is null.
*/
@ParameterizedTest
@MethodSource("executors")
void testExecuteNull(ExecutorService executor) {
try (executor) {
Runnable nullTask = null;
assertThrows(NullPointerException.class, () -> executor.execute(nullTask));
}
}
}

119
test/jdk/java/util/concurrent/Future/DefaultMethods.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2021, 2022, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2021, 2023, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
@ -25,42 +25,50 @@
* @test
* @summary Test Future's default methods
* @library ../lib
* @run testng DefaultMethods
* @run junit DefaultMethods
*/
import java.util.concurrent.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.stream.Stream;
import static java.util.concurrent.Future.State.*;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.*;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import static org.junit.jupiter.api.Assertions.*;
public class DefaultMethods {
class DefaultMethods {
@DataProvider(name = "executors")
public Object[][] executors() {
return new Object[][] {
// ensures that default implementation is tested
{ new DelegatingExecutorService(Executors.newCachedThreadPool()), },
static Stream<ExecutorService> executors() {
return Stream.of(
// ensures that default close method is tested
new DelegatingExecutorService(Executors.newCachedThreadPool()),
// executors that may return a Future that overrides the methods
{ new ForkJoinPool(), },
{ Executors.newCachedThreadPool(), }
};
// executors that may return a Future that overrides the methods
Executors.newCachedThreadPool(),
Executors.newVirtualThreadPerTaskExecutor(),
new ForkJoinPool()
);
}
/**
* Test methods when the task has not completed.
*/
@Test(dataProvider = "executors")
public void testRunningTask(ExecutorService executor) {
@ParameterizedTest
@MethodSource("executors")
void testRunningTask(ExecutorService executor) {
try (executor) {
var latch = new CountDownLatch(1);
Future<?> future = executor.submit(() -> { latch.await(); return null; });
try {
assertTrue(future.state() == RUNNING);
expectThrows(IllegalStateException.class, future::resultNow);
expectThrows(IllegalStateException.class, future::exceptionNow);
assertThrows(IllegalStateException.class, future::resultNow);
assertThrows(IllegalStateException.class, future::exceptionNow);
} finally {
latch.countDown();
}
@ -70,41 +78,44 @@ public class DefaultMethods {
/**
* Test methods when the task has already completed with a result.
*/
@Test(dataProvider = "executors")
public void testCompletedTask1(ExecutorService executor) {
@ParameterizedTest
@MethodSource("executors")
void testCompletedTask1(ExecutorService executor) {
try (executor) {
Future<String> future = executor.submit(() -> "foo");
awaitDone(future);
assertTrue(future.state() == SUCCESS);
assertEquals(future.resultNow(), "foo");
expectThrows(IllegalStateException.class, future::exceptionNow);
assertEquals("foo", future.resultNow());
assertThrows(IllegalStateException.class, future::exceptionNow);
}
}
/**
* Test methods when the task has already completed with null.
*/
@Test(dataProvider = "executors")
public void testCompletedTask2(ExecutorService executor) {
@ParameterizedTest
@MethodSource("executors")
void testCompletedTask2(ExecutorService executor) {
try (executor) {
Future<String> future = executor.submit(() -> null);
awaitDone(future);
assertTrue(future.state() == SUCCESS);
assertEquals(future.resultNow(), null);
expectThrows(IllegalStateException.class, future::exceptionNow);
assertNull(future.resultNow());
assertThrows(IllegalStateException.class, future::exceptionNow);
}
}
/**
* Test methods when the task has completed with an exception.
*/
@Test(dataProvider = "executors")
public void testFailedTask(ExecutorService executor) {
@ParameterizedTest
@MethodSource("executors")
void testFailedTask(ExecutorService executor) {
try (executor) {
Future<?> future = executor.submit(() -> { throw new ArithmeticException(); });
awaitDone(future);
assertTrue(future.state() == FAILED);
expectThrows(IllegalStateException.class, future::resultNow);
assertThrows(IllegalStateException.class, future::resultNow);
Throwable ex = future.exceptionNow();
assertTrue(ex instanceof ArithmeticException);
}
@ -113,16 +124,17 @@ public class DefaultMethods {
/**
* Test methods when the task has been cancelled (mayInterruptIfRunning=false)
*/
@Test(dataProvider = "executors")
public void testCancelledTask1(ExecutorService executor) {
@ParameterizedTest
@MethodSource("executors")
void testCancelledTask1(ExecutorService executor) {
try (executor) {
var latch = new CountDownLatch(1);
Future<?> future = executor.submit(() -> { latch.await(); return null; });
future.cancel(false);
try {
assertTrue(future.state() == CANCELLED);
expectThrows(IllegalStateException.class, future::resultNow);
expectThrows(IllegalStateException.class, future::exceptionNow);
assertThrows(IllegalStateException.class, future::resultNow);
assertThrows(IllegalStateException.class, future::exceptionNow);
} finally {
latch.countDown();
}
@ -132,16 +144,17 @@ public class DefaultMethods {
/**
* Test methods when the task has been cancelled (mayInterruptIfRunning=true)
*/
@Test(dataProvider = "executors")
public void testCancelledTask2(ExecutorService executor) {
@ParameterizedTest
@MethodSource("executors")
void testCancelledTask2(ExecutorService executor) {
try (executor) {
var latch = new CountDownLatch(1);
Future<?> future = executor.submit(() -> { latch.await(); return null; });
future.cancel(true);
try {
assertTrue(future.state() == CANCELLED);
expectThrows(IllegalStateException.class, future::resultNow);
expectThrows(IllegalStateException.class, future::exceptionNow);
assertThrows(IllegalStateException.class, future::resultNow);
assertThrows(IllegalStateException.class, future::exceptionNow);
} finally {
latch.countDown();
}
@ -152,46 +165,46 @@ public class DefaultMethods {
* Test CompletableFuture with the task has not completed.
*/
@Test
public void testCompletableFuture1() {
void testCompletableFuture1() {
var future = new CompletableFuture<String>();
assertTrue(future.state() == RUNNING);
expectThrows(IllegalStateException.class, future::resultNow);
expectThrows(IllegalStateException.class, future::exceptionNow);
assertThrows(IllegalStateException.class, future::resultNow);
assertThrows(IllegalStateException.class, future::exceptionNow);
}
/**
* Test CompletableFuture with the task that completed with result.
*/
@Test
public void testCompletableFuture2() {
void testCompletableFuture2() {
var future = new CompletableFuture<String>();
future.complete("foo");
assertTrue(future.state() == SUCCESS);
assertEquals(future.resultNow(), "foo");
expectThrows(IllegalStateException.class, future::exceptionNow);
assertEquals("foo", future.resultNow());
assertThrows(IllegalStateException.class, future::exceptionNow);
}
/**
* Test CompletableFuture with the task that completed with null.
*/
@Test
public void testCompletableFuture3() {
void testCompletableFuture3() {
var future = new CompletableFuture<String>();
future.complete(null);
assertTrue(future.state() == SUCCESS);
assertEquals(future.resultNow(), null);
expectThrows(IllegalStateException.class, future::exceptionNow);
assertNull(future.resultNow());
assertThrows(IllegalStateException.class, future::exceptionNow);
}
/**
* Test CompletableFuture with the task that completed with exception.
*/
@Test
public void testCompletableFuture4() {
void testCompletableFuture4() {
var future = new CompletableFuture<String>();
future.completeExceptionally(new ArithmeticException());
assertTrue(future.state() == FAILED);
expectThrows(IllegalStateException.class, future::resultNow);
assertThrows(IllegalStateException.class, future::resultNow);
Throwable ex = future.exceptionNow();
assertTrue(ex instanceof ArithmeticException);
}
@ -200,12 +213,12 @@ public class DefaultMethods {
* Test CompletableFuture with the task that was cancelled.
*/
@Test
public void testCompletableFuture5() {
void testCompletableFuture5() {
var future = new CompletableFuture<String>();
future.cancel(false);
assertTrue(future.state() == CANCELLED);
expectThrows(IllegalStateException.class, future::resultNow);
expectThrows(IllegalStateException.class, future::exceptionNow);
assertThrows(IllegalStateException.class, future::resultNow);
assertThrows(IllegalStateException.class, future::exceptionNow);
}
/**

1
test/jdk/java/util/concurrent/TEST.properties Normal file
View File

@ -0,0 +1 @@
maxOutputSize=2000000

95
test/jdk/java/util/concurrent/forkjoin/AsyncShutdownNow.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2020, 2022, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2020, 2023, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
@ -23,11 +23,11 @@
/*
* @test
* @run testng AsyncShutdownNow
* @summary Test invoking shutdownNow with threads blocked in Future.get,
* invokeAll, and invokeAny
* @summary Test ForkJoinPool.shutdownNow with threads blocked in invokeXXX and Future.get
* @run junit AsyncShutdownNow
*/
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
@ -38,42 +38,39 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import static java.lang.Thread.State.*;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.*;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import static org.junit.jupiter.api.Assertions.*;
public class AsyncShutdownNow {
class AsyncShutdownNow {
// long running interruptible task
private static final Callable<Void> SLEEP_FOR_A_DAY = () -> {
Thread.sleep(86400_000);
Thread.sleep(Duration.ofDays(1));
return null;
};
/**
* The executors to test.
*/
@DataProvider(name = "executors")
public Object[][] executors() {
return new Object[][] {
{ new ForkJoinPool() },
{ new ForkJoinPool(1) },
};
static Stream<ForkJoinPool> pools() {
return Stream.of(
new ForkJoinPool(),
new ForkJoinPool(1)
);
}
/**
* Test shutdownNow with running task and thread blocked in Future::get.
* Test shutdownNow with a running task and main thread blocked in Future::get.
*/
@Test(dataProvider = "executors")
public void testFutureGet(ExecutorService executor) throws Exception {
System.out.format("testFutureGet: %s%n", executor);
try (executor) {
Future<?> future = executor.submit(SLEEP_FOR_A_DAY);
@ParameterizedTest
@MethodSource("pools")
void testFutureGet(ForkJoinPool pool) throws Exception {
try (pool) {
Future<?> future = pool.submit(SLEEP_FOR_A_DAY);
// shutdownNow when main thread waits in ForkJoinTask.get
onWait("java.util.concurrent.ForkJoinTask.get", executor::shutdownNow);
// shutdownNow when main thread waits in ForkJoinTask.awaitDone
onWait("java.util.concurrent.ForkJoinTask.awaitDone", pool::shutdownNow);
try {
future.get();
fail();
@ -84,16 +81,16 @@ public class AsyncShutdownNow {
}
/**
* Test shutdownNow with running task and thread blocked in a timed Future::get.
* Test shutdownNow with a running task and main thread blocked in a timed Future::get.
*/
@Test(dataProvider = "executors")
public void testTimedFutureGet(ExecutorService executor) throws Exception {
System.out.format("testTimedFutureGet: %s%n", executor);
try (executor) {
Future<?> future = executor.submit(SLEEP_FOR_A_DAY);
@ParameterizedTest
@MethodSource("pools")
void testTimedFutureGet(ForkJoinPool pool) throws Exception {
try (pool) {
Future<?> future = pool.submit(SLEEP_FOR_A_DAY);
// shutdownNow when main thread waits in ForkJoinTask.get
onWait("java.util.concurrent.ForkJoinTask.get", executor::shutdownNow);
// shutdownNow when main thread waits in ForkJoinTask.awaitDone
onWait("java.util.concurrent.ForkJoinTask.awaitDone", pool::shutdownNow);
try {
future.get(1, TimeUnit.HOURS);
fail();
@ -104,15 +101,15 @@ public class AsyncShutdownNow {
}
/**
* Test shutdownNow with thread blocked in invokeAll.
* Test shutdownNow with running tasks and main thread blocked in invokeAll.
*/
@Test(dataProvider = "executors")
public void testInvokeAll(ExecutorService executor) throws Exception {
System.out.format("testInvokeAll: %s%n", executor);
try (executor) {
// shutdownNow when main thread waits in ForkJoinTask.quietlyJoin
onWait("java.util.concurrent.ForkJoinTask.quietlyJoin", executor::shutdownNow);
List<Future<Void>> futures = executor.invokeAll(List.of(SLEEP_FOR_A_DAY, SLEEP_FOR_A_DAY));
@ParameterizedTest
@MethodSource("pools")
void testInvokeAll(ForkJoinPool pool) throws Exception {
try (pool) {
// shutdownNow when main thread waits in ForkJoinTask.awaitDone
onWait("java.util.concurrent.ForkJoinTask.awaitDone", pool::shutdownNow);
List<Future<Void>> futures = pool.invokeAll(List.of(SLEEP_FOR_A_DAY, SLEEP_FOR_A_DAY));
for (Future<Void> f : futures) {
assertTrue(f.isDone());
try {
@ -126,16 +123,16 @@ public class AsyncShutdownNow {
}
/**
* Test shutdownNow with thread blocked in invokeAny.
* Test shutdownNow with running tasks and main thread blocked in invokeAny.
*/
@Test(dataProvider = "executors", enabled = false)
public void testInvokeAny(ExecutorService executor) throws Exception {
System.out.format("testInvokeAny: %s%n", executor);
try (executor) {
@ParameterizedTest
@MethodSource("pools")
void testInvokeAny(ForkJoinPool pool) throws Exception {
try (pool) {
// shutdownNow when main thread waits in ForkJoinTask.get
onWait("java.util.concurrent.ForkJoinTask.get", executor::shutdownNow);
onWait("java.util.concurrent.ForkJoinTask.get", pool::shutdownNow);
try {
executor.invokeAny(List.of(SLEEP_FOR_A_DAY, SLEEP_FOR_A_DAY));
pool.invokeAny(List.of(SLEEP_FOR_A_DAY, SLEEP_FOR_A_DAY));
fail();
} catch (ExecutionException e) {
// expected

106
test/jdk/java/util/concurrent/tck/ForkJoinPool19Test.java
View File

@ -249,14 +249,17 @@ public class ForkJoinPool19Test extends JSR166TestCase {
FailingFibAction(int n) { number = n; }
public void compute() {
int n = number;
if (n <= 1)
throw new FJException();
else {
FailingFibAction f1 = new FailingFibAction(n - 1);
FailingFibAction f2 = new FailingFibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
if (n > 1) {
try {
FailingFibAction f1 = new FailingFibAction(n - 1);
FailingFibAction f2 = new FailingFibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
return;
} catch (CancellationException fallthrough) {
}
}
throw new FJException();
}
}
@ -389,6 +392,7 @@ public class ForkJoinPool19Test extends JSR166TestCase {
}
f.quietlyJoin();
checkCancelled(f);
Thread.interrupted();
}};
checkInvoke(a);
a.reinitialize();
@ -504,36 +508,78 @@ public class ForkJoinPool19Test extends JSR166TestCase {
* Implicitly closing a new pool using try-with-resources terminates it
*/
public void testClose() {
ForkJoinTask f = new FibAction(8);
ForkJoinPool pool = null;
try (ForkJoinPool p = new ForkJoinPool()) {
pool = p;
p.execute(f);
}
checkCompletedNormally(f);
assertTrue(pool != null && pool.isTerminated());
Thread t = newStartedThread(new CheckedRunnable() {
public void realRun() throws InterruptedException {
FibAction f = new FibAction(1);
ForkJoinPool pool = null;
try (ForkJoinPool p = new ForkJoinPool()) {
pool = p;
p.execute(f);
}
assertTrue(pool != null && pool.isTerminated());
f.join();
assertEquals(1, f.result);
}});
awaitTermination(t);
}
/**
* Explicitly closing a new pool terminates it
*/
public void testClose2() {
Thread t = newStartedThread(new CheckedRunnable() {
public void realRun() throws InterruptedException {
ForkJoinPool pool = new ForkJoinPool();
FibAction f = new FibAction(1);
pool.execute(f);
pool.close();
assertTrue(pool.isTerminated());
f.join();
assertEquals(1, f.result);
}});
awaitTermination(t);
}
/**
* Explicitly closing a shutdown pool awaits termination
*/
public void testClose3() {
Thread t = newStartedThread(new CheckedRunnable() {
public void realRun() throws InterruptedException {
ForkJoinPool pool = new ForkJoinPool();
FibAction f = new FibAction(1);
pool.execute(f);
pool.shutdown();
pool.close();
assertTrue(pool.isTerminated());
f.join();
assertEquals(1, f.result);
}});
awaitTermination(t);
}
/**
* Implicitly closing common pool using try-with-resources has no effect.
*/
public void testCloseCommonPool() {
ForkJoinTask f = new FibAction(8);
ForkJoinPool pool;
try (ForkJoinPool p = pool = ForkJoinPool.commonPool()) {
p.execute(f);
}
assertFalse(pool.isShutdown());
assertFalse(pool.isTerminating());
assertFalse(pool.isTerminated());
String prop = System.getProperty(
"java.util.concurrent.ForkJoinPool.common.parallelism");
if (! "0".equals(prop)) {
f.join();
checkCompletedNormally(f);
}
boolean nothreads = "0".equals(prop);
Thread t = newStartedThread(new CheckedRunnable() {
public void realRun() throws InterruptedException {
ForkJoinTask f = new FibAction(8);
ForkJoinPool pool;
try (ForkJoinPool p = pool = ForkJoinPool.commonPool()) {
p.execute(f);
}
assertFalse(pool.isShutdown());
assertFalse(pool.isTerminating());
assertFalse(pool.isTerminated());
if (!nothreads) {
f.join();
checkCompletedNormally(f);
}
}});
awaitTermination(t);
}
}

21
test/jdk/java/util/concurrent/tck/ForkJoinPool8Test.java
View File

@ -232,14 +232,17 @@ public class ForkJoinPool8Test extends JSR166TestCase {
FailingFibAction(int n) { number = n; }
public void compute() {
int n = number;
if (n <= 1)
throw new FJException();
else {
FailingFibAction f1 = new FailingFibAction(n - 1);
FailingFibAction f2 = new FailingFibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
if (n > 1) {
try {
FailingFibAction f1 = new FailingFibAction(n - 1);
FailingFibAction f2 = new FailingFibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
return;
} catch (CancellationException fallthrough) {
}
}
throw new FJException();
}
}
@ -402,7 +405,9 @@ public class ForkJoinPool8Test extends JSR166TestCase {
try {
f.get(randomTimeout(), null);
shouldThrow();
} catch (NullPointerException success) {}
} catch (NullPointerException success) {
f.join();
}
}};
checkInvoke(a);
}

8
test/jdk/java/util/concurrent/tck/ForkJoinTaskTest.java
View File

@ -496,7 +496,9 @@ public class ForkJoinTaskTest extends JSR166TestCase {
try {
f.get(randomTimeout(), null);
shouldThrow();
} catch (NullPointerException success) {}
} catch (NullPointerException success) {
f.join();
}
}};
testInvokeOnPool(mainPool(), a);
}
@ -1245,7 +1247,9 @@ public class ForkJoinTaskTest extends JSR166TestCase {
try {
f.get(randomTimeout(), null);
shouldThrow();
} catch (NullPointerException success) {}
} catch (NullPointerException success) {
f.join();
}
}};
testInvokeOnPool(singletonPool(), a);
}

26
test/jdk/java/util/concurrent/tck/JSR166TestCase.java
View File

@ -129,6 +129,7 @@ import java.util.Queue;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
@ -1669,11 +1670,20 @@ public class JSR166TestCase extends TestCase {
checkTimedGet(f, expectedValue, LONG_DELAY_MS);
}
// Avoids unwanted interrupts when run inder jtreg
static ThreadGroup topThreadGroup() {
for (ThreadGroup g = Thread.currentThread().getThreadGroup(), p; ; g = p)
if ((p = g.getParent()) == null)
return g;
}
static final ThreadGroup jsr166TestThreadGroup =
new ThreadGroup(topThreadGroup(), "jsr1666TestThreadGroup");
/**
* Returns a new started daemon Thread running the given runnable.
*/
Thread newStartedThread(Runnable runnable) {
Thread t = new Thread(runnable);
Thread t = new Thread(jsr166TestThreadGroup, runnable);
t.setDaemon(true);
t.start();
return t;
@ -1693,10 +1703,12 @@ public class JSR166TestCase extends TestCase {
* the thread (in the hope that it may terminate later) and fails.
*/
void awaitTermination(Thread thread, long timeoutMillis) {
try {
thread.join(timeoutMillis);
} catch (InterruptedException fail) {
threadUnexpectedException(fail);
for (;;) { // ignore stray interrupts by test harness
try {
thread.join(timeoutMillis);
break;
} catch (InterruptedException ignore) {
}
}
if (thread.getState() != Thread.State.TERMINATED) {
String detail = String.format(
@ -1940,6 +1952,8 @@ public class JSR166TestCase extends TestCase {
@Override protected final void compute() {
try {
realCompute();
} catch (CancellationException ex) {
throw ex; // expected by some tests
} catch (Throwable fail) {
threadUnexpectedException(fail);
}
@ -1955,6 +1969,8 @@ public class JSR166TestCase extends TestCase {
@Override protected final T compute() {
try {
return realCompute();
} catch (CancellationException ex) {
throw ex;
} catch (Throwable fail) {
threadUnexpectedException(fail);
}

25
test/jdk/java/util/concurrent/tck/RecursiveActionTest.java
View File

@ -162,10 +162,10 @@ public class RecursiveActionTest extends JSR166TestCase {
void checkCompletedAbnormally(RecursiveAction a, Throwable t) {
assertTrue(a.isDone());
assertFalse(a.isCancelled());
assertFalse(a.isCompletedNormally());
assertTrue(a.isCompletedAbnormally());
assertSame(t.getClass(), a.getException().getClass());
if (!a.isCancelled())
assertSame(t.getClass(), a.getException().getClass());
assertNull(a.getRawResult());
assertFalse(a.cancel(false));
assertFalse(a.cancel(true));
@ -222,14 +222,17 @@ public class RecursiveActionTest extends JSR166TestCase {
FailingFibAction(int n) { number = n; }
public void compute() {
int n = number;
if (n <= 1)
throw new FJException();
else {
FailingFibAction f1 = new FailingFibAction(n - 1);
FailingFibAction f2 = new FailingFibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
if (n > 1) {
try {
FailingFibAction f1 = new FailingFibAction(n - 1);
FailingFibAction f2 = new FailingFibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
return;
} catch (CancellationException fallthrough) {
}
}
throw new FJException();
}
}
@ -488,7 +491,9 @@ public class RecursiveActionTest extends JSR166TestCase {
try {
f.get(randomTimeout(), null);
shouldThrow();
} catch (NullPointerException success) {}
} catch (NullPointerException success) {
f.join();
}
}};
testInvokeOnPool(mainPool(), a);
}

18
test/jdk/java/util/concurrent/tck/RecursiveTaskTest.java
View File

@ -178,10 +178,10 @@ public class RecursiveTaskTest extends JSR166TestCase {
void checkCompletedAbnormally(RecursiveTask<?> a, Throwable t) {
assertTrue(a.isDone());
assertFalse(a.isCancelled());
assertFalse(a.isCompletedNormally());
assertTrue(a.isCompletedAbnormally());
assertSame(t.getClass(), a.getException().getClass());
if (!a.isCancelled())
assertSame(t.getClass(), a.getException().getClass());
assertNull(a.getRawResult());
assertFalse(a.cancel(false));
assertFalse(a.cancel(true));
@ -240,11 +240,15 @@ public class RecursiveTaskTest extends JSR166TestCase {
FailingFibTask(int n) { number = n; }
public Integer compute() {
int n = number;
if (n <= 1)
throw new FJException();
FailingFibTask f1 = new FailingFibTask(n - 1);
f1.fork();
return new FibTask(n - 2).compute() + f1.join();
if (n > 1) {
try {
FailingFibTask f1 = new FailingFibTask(n - 1);
f1.fork();
return new FibTask(n - 2).compute() + f1.join();
} catch (CancellationException fallthrough) {
}
}
throw new FJException();
}
}

1
test/jdk/java/util/concurrent/tck/tck.policy
View File

@ -1,6 +1,7 @@
grant {
// Permissions j.u.c. needs directly
permission java.lang.RuntimePermission "modifyThread";
permission java.lang.RuntimePermission "modifyThreadGroup";
permission java.lang.RuntimePermission "getClassLoader";
permission java.lang.RuntimePermission "setContextClassLoader";
permission java.util.PropertyPermission "*", "read";