Add structured-concurrency-final module: StructuredTaskScope final in JDK 28 (JEP 543), run on 28-ea, 27, 26 and 25
Runnable demos for the ankurm.com structured-concurrency-java article: the ExecutorService thread leak and its fix, every built-in Joiner, structure violations, timeouts, a custom Joiner, JSON thread dumps (in-process and jcmd), scoped values, javap diffs of the API on 25, 26, 27 and 28-ea, compiler output for six pieces of 25/26-era code, and the preview class-file trap between 27 and 28. Transcripts in output/. Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01CifR2BE7PBvRy5hT8TS9oz
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Joiner;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
/** A custom completion policy: keep what succeeds, ignore what fails. The final Joiner has THREE type parameters. */
|
||||
public class CollectingJoiner<T> implements Joiner<T, List<T>, CompletionException> {
|
||||
private final Queue<T> results = new ConcurrentLinkedQueue<>();
|
||||
|
||||
@Override
|
||||
public boolean onComplete(Subtask<T> subtask) {
|
||||
if (subtask.state() == Subtask.State.SUCCESS) {
|
||||
results.add(subtask.get());
|
||||
}
|
||||
return false; // never cancel early: let everyone finish
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<T> result() {
|
||||
return List.copyOf(results);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<T> timeout() {
|
||||
// the scope was opened with a timeout and it expired
|
||||
throw new CompletionException(new StructuredTaskScope.CancelledByTimeoutException());
|
||||
}
|
||||
|
||||
static <T> List<T> allSuccessful(List<Callable<T>> tasks) throws InterruptedException {
|
||||
try (var scope = StructuredTaskScope.open(new CollectingJoiner<T>())) {
|
||||
tasks.forEach(scope::fork);
|
||||
return scope.join();
|
||||
}
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
List<Callable<Integer>> tasks = List.of(
|
||||
() -> 1,
|
||||
() -> { throw new IllegalStateException("ignored"); },
|
||||
() -> 3);
|
||||
List<Integer> r = new ArrayList<>(allSuccessful(tasks));
|
||||
Collections.sort(r);
|
||||
System.out.println("collected (sorted): " + r);
|
||||
|
||||
System.out.println("-- with a 100 ms timeout and a 5 s subtask");
|
||||
try (var scope = StructuredTaskScope.open(new CollectingJoiner<Integer>(),
|
||||
cf -> cf.withTimeout(java.time.Duration.ofMillis(100)))) {
|
||||
scope.fork(() -> { Thread.sleep(5000); return 1; });
|
||||
scope.join();
|
||||
} catch (CompletionException e) {
|
||||
System.out.println(" " + e.getClass().getSimpleName() + ", cause = " + e.getCause().getClass().getName());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
/** The same two subtasks in a StructuredTaskScope: the failure of one cancels the other, before handle() returns. */
|
||||
public class FailFast {
|
||||
|
||||
static String findUser() throws InterruptedException {
|
||||
Thread.sleep(50);
|
||||
throw new IllegalStateException("user service down");
|
||||
}
|
||||
|
||||
static int fetchOrder() throws InterruptedException {
|
||||
try {
|
||||
Thread.sleep(300);
|
||||
System.out.println(" fetchOrder() finished");
|
||||
return 42;
|
||||
} catch (InterruptedException e) {
|
||||
System.out.println(" fetchOrder() was interrupted");
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
static String handle() throws InterruptedException {
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
Subtask<String> user = scope.fork(FailFast::findUser);
|
||||
Subtask<Integer> order = scope.fork(FailFast::fetchOrder);
|
||||
scope.join();
|
||||
return user.get() + " " + order.get();
|
||||
} catch (ExecutionException e) {
|
||||
System.out.println("handle() failed: " + e.getCause());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
handle();
|
||||
System.out.println("handle() has returned. Waiting 400 ms to see what the other thread does:");
|
||||
Thread.sleep(400);
|
||||
System.out.println("(nothing more happened)");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
/** What goes wrong when the structure is violated, and what the exception looks like. */
|
||||
public class Failures {
|
||||
|
||||
static void attempt(String title, Callable<Object> body) {
|
||||
System.out.println("-- " + title);
|
||||
try {
|
||||
Object r = body.call();
|
||||
System.out.println(" returned " + r);
|
||||
} catch (Throwable t) {
|
||||
System.out.println(" " + t.getClass().getName() + ": " + t.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
attempt("Subtask.get() before join()", () -> {
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
Subtask<String> s = scope.fork(() -> "x");
|
||||
return s.get();
|
||||
}
|
||||
});
|
||||
|
||||
attempt("leaving the try block without join()", () -> {
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
scope.fork(() -> "x");
|
||||
return "left the block";
|
||||
}
|
||||
});
|
||||
|
||||
attempt("fork() from a thread that does not own the scope", () -> {
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
var other = new CompletableFuture<Object>();
|
||||
Thread.ofVirtual().start(() -> {
|
||||
try { scope.fork(() -> "x"); other.complete("forked"); }
|
||||
catch (Throwable t) { other.complete(t.getClass().getName() + ": " + t.getMessage()); }
|
||||
});
|
||||
Object r = other.get();
|
||||
scope.join();
|
||||
return r;
|
||||
}
|
||||
});
|
||||
|
||||
attempt("closing scopes out of order (not try-with-resources)", () -> {
|
||||
var outer = StructuredTaskScope.open();
|
||||
var inner = StructuredTaskScope.open();
|
||||
try {
|
||||
outer.close(); // the inner scope is still open
|
||||
return "closed outer first";
|
||||
} finally {
|
||||
try { inner.close(); } catch (Throwable ignore) { }
|
||||
try { outer.close(); } catch (Throwable ignore) { }
|
||||
}
|
||||
});
|
||||
|
||||
attempt("join() twice", () -> {
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
scope.fork(() -> "x");
|
||||
scope.join();
|
||||
return scope.join();
|
||||
}
|
||||
});
|
||||
|
||||
System.out.println("-- the failure surfaces as ExecutionException; pattern-match the cause");
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
scope.fork(() -> { throw new IOException("disk gone"); });
|
||||
scope.join();
|
||||
} catch (ExecutionException e) {
|
||||
Throwable cause = e.getCause();
|
||||
switch (cause) {
|
||||
case IOException ioe -> System.out.println(" I/O failed: " + ioe.getMessage());
|
||||
default -> System.out.println(" unexpected: " + cause);
|
||||
}
|
||||
}
|
||||
|
||||
System.out.println("-- two subtasks fail: which cause do you get?");
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
scope.fork(() -> { Thread.sleep(20); throw new IllegalStateException("second"); });
|
||||
scope.fork(() -> { throw new IllegalStateException("first"); });
|
||||
scope.join();
|
||||
} catch (ExecutionException e) {
|
||||
System.out.println(" cause = " + e.getCause().getMessage() + " (the first to fail wins)");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Joiner;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/** The built-in Joiner factories, each with a result and a failure. */
|
||||
public class Joiners {
|
||||
|
||||
/** Names of the subtasks that were interrupted; printed after the scope closes so the order is stable. */
|
||||
static final Set<String> interrupted = new ConcurrentSkipListSet<>();
|
||||
|
||||
static void reportInterrupted() {
|
||||
System.out.println(" interrupted: " + interrupted);
|
||||
interrupted.clear();
|
||||
}
|
||||
|
||||
static Callable<String> task(String name, long ms) {
|
||||
return () -> {
|
||||
try { Thread.sleep(ms); return name; }
|
||||
catch (InterruptedException e) { interrupted.add(name); throw e; }
|
||||
};
|
||||
}
|
||||
static Callable<String> failing(String name, long ms) {
|
||||
return () -> { Thread.sleep(ms); throw new IllegalStateException(name + " failed"); };
|
||||
}
|
||||
|
||||
/** A checked exception of our own, to show the Function overloads. */
|
||||
static class ServiceException extends Exception {
|
||||
ServiceException(Throwable cause) { super("service call failed: " + cause.getMessage(), cause); }
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
System.out.println("-- allSuccessfulOrThrow(): List in FORK order, not completion order");
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow())) {
|
||||
scope.fork(task("slow", 150)); scope.fork(task("fast", 10)); scope.fork(task("medium", 80));
|
||||
System.out.println(" " + scope.join());
|
||||
}
|
||||
|
||||
System.out.println("-- allSuccessfulOrThrow(): one failure cancels the others, join() throws ExecutionException");
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow())) {
|
||||
scope.fork(task("slow", 500)); scope.fork(failing("broken", 30));
|
||||
scope.join();
|
||||
} catch (ExecutionException e) {
|
||||
System.out.println(" " + e.getClass().getSimpleName() + ", cause = " + e.getCause());
|
||||
}
|
||||
reportInterrupted();
|
||||
|
||||
System.out.println("-- anySuccessfulOrThrow(): the race; losers are interrupted");
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulOrThrow())) {
|
||||
scope.fork(task("A-300ms", 300)); scope.fork(task("B-50ms", 50)); scope.fork(task("C-300ms", 300));
|
||||
System.out.println(" winner: " + scope.join());
|
||||
}
|
||||
reportInterrupted();
|
||||
|
||||
System.out.println("-- anySuccessfulOrThrow(): when every subtask fails");
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulOrThrow())) {
|
||||
scope.fork(failing("one", 10)); scope.fork(failing("two", 20));
|
||||
scope.join();
|
||||
} catch (ExecutionException e) {
|
||||
System.out.println(" " + e.getClass().getSimpleName() + ", cause = " + e.getCause());
|
||||
}
|
||||
|
||||
System.out.println("-- awaitAllSuccessfulOrThrow(): returns null (Void), like the zero-argument open()");
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow())) {
|
||||
Subtask<String> a = scope.fork(task("a", 10));
|
||||
Subtask<String> b = scope.fork(task("b", 20));
|
||||
Void v = scope.join();
|
||||
System.out.println(" join() = " + v + ", a=" + a.get() + ", b=" + b.get());
|
||||
}
|
||||
|
||||
System.out.println("-- allUntil(predicate): stop when two subtasks have completed");
|
||||
var done = new AtomicInteger();
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>allUntil(s -> done.incrementAndGet() == 2))) {
|
||||
scope.fork(task("t1", 20)); scope.fork(task("t2", 60)); scope.fork(task("t3", 400)); scope.fork(task("t4", 400));
|
||||
List<Subtask<String>> subtasks = scope.join();
|
||||
subtasks.forEach(s -> System.out.println(" " + s.state() + (s.state() == Subtask.State.SUCCESS ? " " + s.get() : "")));
|
||||
}
|
||||
reportInterrupted();
|
||||
|
||||
System.out.println("-- the Function overload: join() throws OUR exception type");
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String, ServiceException>allSuccessfulOrThrow(ServiceException::new))) {
|
||||
scope.fork(failing("orders", 10));
|
||||
scope.join();
|
||||
} catch (ServiceException e) {
|
||||
System.out.println(" caught " + e.getClass().getSimpleName() + ": " + e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
import java.time.Duration;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Joiner;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
/** One line per API in the article's quick reference. Compiles on JDK 28 with no flags. */
|
||||
public class QuickReference {
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
static void shapes(ThreadFactory factory) throws Exception {
|
||||
// region quickref
|
||||
// Open a scope: the policy is the Joiner, the type parameters are <T subtask result, R join() result, R_X join() exception>
|
||||
var plain = StructuredTaskScope.<String>open(); // join() -> Void, throws ExecutionException
|
||||
var all = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow()); // join() -> List<String>, in fork order
|
||||
var any = StructuredTaskScope.open(Joiner.<String>anySuccessfulOrThrow()); // join() -> String, first success wins
|
||||
var await = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow()); // join() -> Void
|
||||
var until = StructuredTaskScope.open(Joiner.<String>allUntil(s -> false)); // join() -> List<Subtask<String>>, no checked exception
|
||||
var mine = StructuredTaskScope.open(Joiner.<String, IllegalStateException>allSuccessfulOrThrow(IllegalStateException::new)); // your exception type
|
||||
|
||||
// Configure: name for thread dumps, a deadline for the whole group, a thread factory for the subtasks
|
||||
var cfg = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(),
|
||||
cf -> cf.withName("orders").withTimeout(Duration.ofMillis(500)).withThreadFactory(factory));
|
||||
|
||||
// Fork, then read each Subtask only after join()
|
||||
Subtask<String> subtask = all.fork(() -> "value");
|
||||
all.join();
|
||||
String value = subtask.get(); // like Future.resultNow()
|
||||
Subtask.State state = subtask.state(); // SUCCESS | FAILED | UNAVAILABLE
|
||||
// Subtask also has exception(); see output/08-api-28.txt for every signature
|
||||
// endregion quickref
|
||||
// scopes opened by one thread must be closed in the reverse order (try-with-resources does this for you)
|
||||
cfg.close(); mine.close(); until.close(); await.close(); any.close(); all.close(); plain.close();
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
shapes(Thread.ofVirtual().factory());
|
||||
System.out.println("quick reference compiled and ran");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Joiner;
|
||||
|
||||
/** Five tasks that sleep a random time and fail if they were "too slow". Seeded, so the transcript is repeatable. */
|
||||
public class RandomTasks {
|
||||
|
||||
static class TooSlowException extends Exception {
|
||||
TooSlowException(String m) { super(m); }
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
long seed = Long.parseLong(args[0]);
|
||||
int threshold = 700;
|
||||
Random random = new Random(seed);
|
||||
List<Callable<Integer>> tasks = new ArrayList<>();
|
||||
for (int i = 0; i < 5; i++) {
|
||||
int duration = random.nextInt(1000);
|
||||
System.out.println("Duration: " + duration);
|
||||
tasks.add(() -> {
|
||||
Thread.sleep(duration);
|
||||
if (duration > threshold) throw new TooSlowException("Duration " + duration + " greater than threshold " + threshold);
|
||||
return duration;
|
||||
});
|
||||
}
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<Integer>allSuccessfulOrThrow())) {
|
||||
tasks.forEach(scope::fork);
|
||||
for (int r : scope.join()) System.out.println("Result: " + r);
|
||||
} catch (ExecutionException e) {
|
||||
System.out.println("ExecutionException: " + e.getCause().getClass().getSimpleName() + ": " + e.getCause().getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
/** Subtasks inherit ScopedValue bindings (JEP 506, final in JDK 25). A plain thread does not. */
|
||||
public class ScopedValues {
|
||||
static final ScopedValue<String> REQUEST_ID = ScopedValue.newInstance();
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
ScopedValue.where(REQUEST_ID, "req-7731").run(() -> {
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
Subtask<String> a = scope.fork(() -> "forked subtask sees " + REQUEST_ID.get());
|
||||
scope.join();
|
||||
System.out.println(a.get());
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
Thread t = new Thread(() ->
|
||||
System.out.println("plain thread: bound? " + REQUEST_ID.isBound()));
|
||||
t.start();
|
||||
try { t.join(); } catch (InterruptedException e) { }
|
||||
});
|
||||
System.out.println("outside the binding: bound? " + REQUEST_ID.isBound());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.StructuredTaskScope;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
record Response(String user, int order) {}
|
||||
|
||||
/** The canonical program: open, fork, join, process, close. */
|
||||
public class Server {
|
||||
|
||||
String findUser() throws InterruptedException {
|
||||
Thread.sleep(120); // pretend: a network call
|
||||
return "Alice";
|
||||
}
|
||||
|
||||
int fetchOrder() throws InterruptedException {
|
||||
Thread.sleep(80); // pretend: another network call
|
||||
return 42;
|
||||
}
|
||||
|
||||
// join() now declares ExecutionException (the scope's third type parameter, R_X) as well as InterruptedException
|
||||
Response handle() throws ExecutionException, InterruptedException {
|
||||
try (var scope = StructuredTaskScope.open()) { // 1. open
|
||||
Subtask<String> user = scope.fork(this::findUser); // 2. fork
|
||||
Subtask<Integer> order = scope.fork(this::fetchOrder); // fork
|
||||
|
||||
scope.join(); // 3. wait for BOTH, propagate any failure
|
||||
|
||||
return new Response(user.get(), order.get()); // 4. process results
|
||||
} // 5. close: cancel + wait (automatic)
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
System.out.println(new Server().handle());
|
||||
System.out.println("Runtime: " + Runtime.version());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
import java.io.IOException;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Joiner;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
/** The short fragments quoted in the article, as one compiling file (see output/12-snippets-compile.txt). */
|
||||
public class Snippets {
|
||||
record Response(String user, int order) {}
|
||||
|
||||
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
String findUser() throws InterruptedException { return "Alice"; }
|
||||
int fetchOrder() throws InterruptedException { return 42; }
|
||||
|
||||
class ExecutorVersion {
|
||||
// region unstructured
|
||||
Response handle() throws ExecutionException, InterruptedException {
|
||||
Future<String> user = executor.submit(() -> findUser());
|
||||
Future<Integer> order = executor.submit(() -> fetchOrder());
|
||||
|
||||
String theUser = user.get(); // join findUser
|
||||
int theOrder = order.get(); // join fetchOrder
|
||||
return new Response(theUser, theOrder);
|
||||
}
|
||||
// endregion unstructured
|
||||
}
|
||||
|
||||
class SequentialVersion {
|
||||
// region sequential
|
||||
Response handle() throws IOException, InterruptedException {
|
||||
String theUser = findUser();
|
||||
int theOrder = fetchOrder();
|
||||
return new Response(theUser, theOrder);
|
||||
}
|
||||
// endregion sequential
|
||||
}
|
||||
|
||||
// region runConcurrently
|
||||
<T> List<T> runConcurrently(Collection<Callable<T>> tasks) throws ExecutionException, InterruptedException {
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<T>allSuccessfulOrThrow())) {
|
||||
tasks.forEach(scope::fork);
|
||||
return scope.join(); // List<T> of results; throws ExecutionException if any task fails
|
||||
}
|
||||
}
|
||||
// endregion runConcurrently
|
||||
|
||||
// region race
|
||||
<T> T race(Collection<Callable<T>> tasks) throws ExecutionException, InterruptedException {
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<T>anySuccessfulOrThrow())) {
|
||||
tasks.forEach(scope::fork);
|
||||
return scope.join(); // result of the first subtask to succeed
|
||||
}
|
||||
}
|
||||
// endregion race
|
||||
|
||||
// region handleWithCatch
|
||||
Response handleWithCatch() throws InterruptedException {
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
Subtask<String> user = scope.fork(this::findUser);
|
||||
Subtask<Integer> order = scope.fork(this::fetchOrder);
|
||||
scope.join();
|
||||
return new Response(user.get(), order.get());
|
||||
} catch (ExecutionException e) {
|
||||
Throwable cause = e.getCause();
|
||||
switch (cause) {
|
||||
case IOException ioe -> System.err.println("I/O failed: " + ioe);
|
||||
default -> System.err.println("Unexpected: " + cause);
|
||||
}
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
// endregion handleWithCatch
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
var s = new Snippets();
|
||||
try {
|
||||
System.out.println(s.new ExecutorVersion().handle());
|
||||
System.out.println(s.new SequentialVersion().handle());
|
||||
System.out.println(s.runConcurrently(List.of(() -> 1, () -> 2)));
|
||||
String winner = s.race(List.<Callable<String>>of(() -> "a")); System.out.println(winner);
|
||||
System.out.println(s.handleWithCatch());
|
||||
} finally {
|
||||
s.executor.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
import java.util.concurrent.*;
|
||||
|
||||
/** A subtask that ignores interrupts keeps close() waiting, even though the scope was cancelled at once. */
|
||||
public class StuckClose {
|
||||
|
||||
static int stubborn() {
|
||||
long end = System.nanoTime() + 1_500_000_000L;
|
||||
while (System.nanoTime() < end) { /* busy loop: never checks the interrupt flag */ }
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int cooperative() throws InterruptedException {
|
||||
Thread.sleep(1_500); // blocking calls like sleep() respond to interrupt
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void run(String label, Callable<Integer> slow) throws Exception {
|
||||
long t0 = System.nanoTime();
|
||||
try (var scope = StructuredTaskScope.open()) {
|
||||
scope.fork(slow);
|
||||
scope.fork(() -> { Thread.sleep(50); throw new IllegalStateException("fails at 50 ms"); });
|
||||
scope.join();
|
||||
} catch (ExecutionException e) {
|
||||
long ms = (System.nanoTime() - t0) / 1_000_000;
|
||||
System.out.println(label + ": caught " + e.getCause().getMessage() + "; the try block took >= 1400 ms: " + (ms >= 1400));
|
||||
}
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
run("subtask that responds to interrupts", StuckClose::cooperative);
|
||||
run("subtask that ignores interrupts ", StuckClose::stubborn);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
import com.sun.management.HotSpotDiagnosticMXBean;
|
||||
import java.lang.management.ManagementFactory;
|
||||
import java.nio.file.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Joiner;
|
||||
|
||||
/** The thread dump shows the scope as a container, its owner, and the threads forked in it. */
|
||||
public class ThreadDump {
|
||||
public static void main(String[] args) throws Exception {
|
||||
if (args.length > 0 && args[0].equals("hold")) { // used by run.sh: hold nested scopes open so jcmd can dump them
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow(), cf -> cf.withName("orders"))) {
|
||||
scope.fork(() -> { Thread.sleep(6000); return "A"; });
|
||||
scope.fork(() -> { // subtask B opens a child scope of its own
|
||||
try (var child = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow(), cf -> cf.withName("inventory"))) {
|
||||
child.fork(() -> { Thread.sleep(6000); return "B1"; });
|
||||
child.join();
|
||||
}
|
||||
return "B";
|
||||
});
|
||||
scope.join();
|
||||
}
|
||||
return;
|
||||
}
|
||||
Path file = Files.createTempFile("dump", ".json"); Files.delete(file);
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow(), cf -> cf.withName("orders"))) {
|
||||
for (int i = 0; i < 2; i++) {
|
||||
int n = i;
|
||||
scope.fork(() -> { Thread.sleep(1500); return "task-" + n; });
|
||||
}
|
||||
Thread.sleep(300); // let the subtasks start and block
|
||||
ManagementFactory.getPlatformMXBean(HotSpotDiagnosticMXBean.class)
|
||||
.dumpThreads(file.toString(), HotSpotDiagnosticMXBean.ThreadDumpFormat.JSON);
|
||||
scope.join();
|
||||
}
|
||||
String json = Files.readString(file);
|
||||
int at = json.indexOf("\"container\": \"orders");
|
||||
int start = json.lastIndexOf('{', at);
|
||||
int end = json.indexOf("\"threadCount\"", at);
|
||||
String block = json.substring(start, end);
|
||||
// identity hashes, thread ids and timestamps change between runs: mask them so the transcript is stable
|
||||
block = block.replaceAll("@[0-9a-f]+", "@<hash>").replaceAll("\"tid\": \\d+", "\"tid\": <n>").replaceAll("\"owner\": \\d+", "\"owner\": <n>");
|
||||
// keep the lines that show the hierarchy: the scope, its parent and owner, and each forked thread with its own frame
|
||||
for (String line : block.split("\n")) {
|
||||
String l = line.trim();
|
||||
if (l.startsWith("\"container\"") || l.startsWith("\"parent\"") || l.startsWith("\"owner\"")
|
||||
|| l.startsWith("\"tid\"") || l.startsWith("\"virtual\"") || l.startsWith("\"state\"")
|
||||
|| l.contains("ThreadDump.lambda")) {
|
||||
System.out.println(l);
|
||||
}
|
||||
}
|
||||
Files.deleteIfExists(file);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
import java.time.Duration;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.StructuredTaskScope.Joiner;
|
||||
import java.util.concurrent.StructuredTaskScope.Subtask;
|
||||
|
||||
/** Deadlines for a whole group of subtasks, and what the Joiner's timeout() method does with them. */
|
||||
public class Timeouts {
|
||||
|
||||
static <T> List<T> runConcurrently(Collection<Callable<T>> tasks, ThreadFactory factory, Duration timeout)
|
||||
throws ExecutionException, InterruptedException {
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<T>allSuccessfulOrThrow(),
|
||||
cf -> cf.withName("orders").withThreadFactory(factory).withTimeout(timeout))) {
|
||||
tasks.forEach(scope::fork);
|
||||
return scope.join();
|
||||
}
|
||||
}
|
||||
|
||||
/** Collects successes; on timeout it hands back what it has instead of failing. */
|
||||
static class BestEffort<T> implements Joiner<T, List<T>, RuntimeException> {
|
||||
private final Queue<T> results = new ConcurrentLinkedQueue<>();
|
||||
public boolean onComplete(Subtask<T> s) {
|
||||
if (s.state() == Subtask.State.SUCCESS) results.add(s.get());
|
||||
return false;
|
||||
}
|
||||
public List<T> result() { return List.copyOf(results); }
|
||||
public List<T> timeout() { return List.copyOf(results); } // no exception: a partial answer
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
ThreadFactory factory = Thread.ofVirtual().name("order-", 0).factory();
|
||||
Callable<String> fast = () -> { Thread.sleep(20); return "fast"; };
|
||||
Callable<String> slow = () -> { Thread.sleep(5000); return "slow"; };
|
||||
|
||||
System.out.println("-- inside the deadline");
|
||||
System.out.println(" " + runConcurrently(List.of(fast, fast), factory, Duration.ofSeconds(1)));
|
||||
|
||||
System.out.println("-- deadline 200 ms, one subtask needs 5 s");
|
||||
long t0 = System.nanoTime();
|
||||
try {
|
||||
runConcurrently(List.of(fast, slow), factory, Duration.ofMillis(200));
|
||||
} catch (ExecutionException e) {
|
||||
long ms = (System.nanoTime() - t0) / 1_000_000;
|
||||
System.out.println(" " + e.getClass().getSimpleName() + ", cause = " + e.getCause().getClass().getName());
|
||||
System.out.println(" returned within " + (ms < 1000 ? "well under a second" : ms + " ms") + " (not 5 s): the slow subtask was interrupted");
|
||||
}
|
||||
|
||||
System.out.println("-- the same deadline with a joiner whose timeout() returns a partial result");
|
||||
try (var scope = StructuredTaskScope.open(new BestEffort<String>(), cf -> cf.withTimeout(Duration.ofMillis(200)))) {
|
||||
scope.fork(fast); scope.fork(slow);
|
||||
System.out.println(" join() = " + scope.join());
|
||||
}
|
||||
|
||||
System.out.println("-- a thread factory decides the subtask thread's name");
|
||||
try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(), cf -> cf.withThreadFactory(Thread.ofVirtual().name("order-", 0).factory()))) {
|
||||
scope.fork(() -> Thread.currentThread().getName());
|
||||
scope.fork(() -> Thread.currentThread().getName());
|
||||
System.out.println(" " + scope.join());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
import java.util.concurrent.*;
|
||||
|
||||
/** The ExecutorService version of handle(): findUser() fails, and fetchOrder() keeps running. */
|
||||
public class Unstructured {
|
||||
|
||||
static String findUser() throws InterruptedException {
|
||||
Thread.sleep(50);
|
||||
throw new IllegalStateException("user service down");
|
||||
}
|
||||
|
||||
static int fetchOrder() throws InterruptedException {
|
||||
try {
|
||||
Thread.sleep(300);
|
||||
System.out.println(" fetchOrder() finished, although nobody wants the answer any more");
|
||||
return 42;
|
||||
} catch (InterruptedException e) {
|
||||
System.out.println(" fetchOrder() was interrupted");
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
try {
|
||||
Future<String> user = executor.submit(Unstructured::findUser);
|
||||
Future<Integer> order = executor.submit(Unstructured::fetchOrder);
|
||||
try {
|
||||
String u = user.get(); // throws after ~50 ms ...
|
||||
int o = order.get();
|
||||
System.out.println(u + " " + o);
|
||||
} catch (ExecutionException e) {
|
||||
System.out.println("handle() failed: " + e.getCause());
|
||||
}
|
||||
System.out.println("handle() has returned. Waiting 400 ms to see what the other thread does:");
|
||||
Thread.sleep(400);
|
||||
} finally {
|
||||
executor.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user