You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@groovy.apache.org by su...@apache.org on 2026/07/20 18:29:10 UTC
(groovy) branch GROOVY-12181 updated (305bc11cc3 -> 9d50409a09)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch GROOVY-12181
in repository https://gitbox.apache.org/repos/asf/groovy.git
discard 305bc11cc3 GROOVY-12181: Refactor async runtime and AST helpers introduced by GROOVY-9381
new 9d50409a09 GROOVY-12181: Refactor async runtime and AST helpers introduced by GROOVY-9381
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (305bc11cc3)
\
N -- N -- N refs/heads/GROOVY-12181 (9d50409a09)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/groovy/runtime/async/AsyncSupport.java | 25 +++++-----------------
.../groovy/transform/AsyncTransformHelper.java | 19 +++++++++++-----
src/test/groovy/groovy/AsyncAwaitTest.groovy | 9 ++++----
.../runtime/async/AsyncSupportInternalsTest.groovy | 10 ++++-----
.../transform/AsyncTransformHelperTest.groovy | 9 ++++++--
5 files changed, 36 insertions(+), 36 deletions(-)
(groovy) 01/01: GROOVY-12181: Refactor async runtime and AST helpers introduced by GROOVY-9381
Posted by su...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch GROOVY-12181
in repository https://gitbox.apache.org/repos/asf/groovy.git
commit 9d50409a09cecb77f4d47589f50d2965073da4f9
Author: Daniel Sun <su...@apache.org>
AuthorDate: Tue Jul 21 00:58:40 2026 +0900
GROOVY-12181: Refactor async runtime and AST helpers introduced by GROOVY-9381
---
src/main/java/groovy/concurrent/AwaitResult.java | 5 +-
src/main/java/groovy/concurrent/Awaitable.java | 7 +-
src/main/java/groovy/concurrent/package-info.java | 17 +-
.../apache/groovy/parser/antlr4/AstBuilder.java | 49 +---
.../groovy/runtime/async/AsyncExecutors.java | 128 ++++++++++
.../apache/groovy/runtime/async/AsyncSupport.java | 282 +++++----------------
.../groovy/runtime/async/AwaitCombinators.java | 201 +++++++++++++++
.../apache/groovy/runtime/async/GroovyPromise.java | 9 +-
.../apache/groovy/runtime/async/package-info.java | 20 +-
.../groovy/transform/AsyncTransformHelper.java | 127 +++++++---
src/spec/doc/core-async-await.adoc | 4 +-
src/test/groovy/groovy/AsyncAwaitTest.groovy | 40 +++
.../groovy/concurrent/AwaitResultTest.groovy | 118 +++++++++
.../runtime/async/AsyncSupportInternalsTest.groovy | 259 +++++++++++++++++++
.../transform/AsyncTransformHelperTest.groovy | 184 ++++++++++++++
15 files changed, 1131 insertions(+), 319 deletions(-)
diff --git a/src/main/java/groovy/concurrent/AwaitResult.java b/src/main/java/groovy/concurrent/AwaitResult.java
index f7f77879b9..eed54a524c 100644
--- a/src/main/java/groovy/concurrent/AwaitResult.java
+++ b/src/main/java/groovy/concurrent/AwaitResult.java
@@ -69,9 +69,8 @@ public final class AwaitResult<T> {
* @param <T> the value type
* @return a success result wrapping the value
*/
- @SuppressWarnings("unchecked")
- public static <T> AwaitResult<T> success(Object value) {
- return new AwaitResult<>((T) value, null, true);
+ public static <T> AwaitResult<T> success(T value) {
+ return new AwaitResult<>(value, null, true);
}
/**
diff --git a/src/main/java/groovy/concurrent/Awaitable.java b/src/main/java/groovy/concurrent/Awaitable.java
index 59f06f0876..bb941cd1e7 100644
--- a/src/main/java/groovy/concurrent/Awaitable.java
+++ b/src/main/java/groovy/concurrent/Awaitable.java
@@ -439,9 +439,10 @@ public interface Awaitable<T> {
* Returns an {@code Awaitable} that completes with the result of the first
* source that succeeds. Individual failures are silently absorbed; only
* when <em>all</em> sources have failed does the returned awaitable reject
- * with an {@link IllegalStateException} whose
- * {@linkplain Throwable#getSuppressed() suppressed} array contains every
- * individual error.
+ * with an aggregate {@link java.util.concurrent.CompletionException}
+ * (message {@code "All N tasks failed"}, cause = first failure, remaining
+ * failures as suppressed). Under {@code await} exception transparency the
+ * cause is rethrown.
* <p>
* This is the Groovy equivalent of JavaScript's {@code Promise.any()}.
* Contrast with {@link #any(Object...)} which returns the first result to
diff --git a/src/main/java/groovy/concurrent/package-info.java b/src/main/java/groovy/concurrent/package-info.java
index 1ffb533d6d..950b890619 100644
--- a/src/main/java/groovy/concurrent/package-info.java
+++ b/src/main/java/groovy/concurrent/package-info.java
@@ -21,10 +21,23 @@
* Structured concurrency abstractions for asynchronous and parallel programming.
*
* <p>
- * Provides high-level concurrency primitives: {@link groovy.concurrent.Actor} (message-passing),
- * {@link groovy.concurrent.AsyncScope} (structured concurrency), {@link groovy.concurrent.Pool} (thread management),
+ * Language-facing async types:
+ * <ul>
+ * <li>{@link groovy.concurrent.Awaitable} — async computation handle and
+ * static combinator surface used with the {@code await} keyword</li>
+ * <li>{@link groovy.concurrent.AwaitResult} — success/failure value from
+ * {@link groovy.concurrent.Awaitable#allSettled(Object...)}</li>
+ * <li>{@link groovy.concurrent.AsyncScope} — structured concurrency</li>
+ * <li>{@link groovy.concurrent.AsyncChannel} — CSP-style async channel</li>
+ * <li>{@link groovy.concurrent.AwaitableAdapter} /
+ * {@link groovy.concurrent.AwaitableAdapterRegistry} — SPI for
+ * third-party async types (RxJava, Reactor, …)</li>
+ * </ul>
+ * Additional high-level primitives: {@link groovy.concurrent.Actor}
+ * (message-passing), {@link groovy.concurrent.Pool} (thread management),
* and {@link groovy.concurrent.ParallelScope} (combined async+parallel).
* Supports virtual threads (JDK 21+) and fail-fast exception handling.
+ * Runtime implementation details live in {@code org.apache.groovy.runtime.async}.
* </p>
*/
package groovy.concurrent;
diff --git a/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java b/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java
index deaaf9988d..a35a61c791 100644
--- a/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java
+++ b/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java
@@ -539,20 +539,7 @@ public class AstBuilder extends GroovyParserBaseVisitor<Object> {
visitAnnotationsOpt(ctx.annotationsOpt()).forEach(forStatement::addStatementAnnotation);
if (isForAwait) {
- // Transform collection expression: wrap in AsyncSupport.toIterable()
- // and wrap the loop in try/finally to ensure cleanup on break/exception
- Expression original = forStatement.getCollectionExpression();
- String tempVar = "__forAwaitSource" + ctx.hashCode();
- Expression toIterableCall = AsyncTransformHelper.buildToIterableCall(original);
-
- // var $temp = AsyncSupport.toIterable(original)
- Statement declStmt = stmt(declX(varX(tempVar), toIterableCall));
- forStatement.setCollectionExpression(varX(tempVar));
-
- // try { for (...) { body } } finally { AsyncSupport.closeIterable($temp) }
- Statement finallyStmt = stmt(AsyncTransformHelper.buildCloseIterableCall(varX(tempVar)));
- TryCatchStatement tryCatch = new TryCatchStatement(forStatement, finallyStmt);
- return configureAST(block(declStmt, tryCatch), ctx);
+ return configureAST(AsyncTransformHelper.wrapForAwaitLoop(forStatement), ctx);
}
return forStatement;
@@ -3090,39 +3077,7 @@ public class AstBuilder extends GroovyParserBaseVisitor<Object> {
@Override
public Expression visitAsyncClosureExprAlt(final AsyncClosureExprAltContext ctx) {
ClosureExpression closure = this.visitClosureOrLambdaExpression(ctx.closureOrLambdaExpression());
- boolean hasYieldReturn = AsyncTransformHelper.containsYieldReturn(closure.getCode());
- boolean hasDefer = AsyncTransformHelper.containsDefer(closure.getCode());
-
- if (hasDefer) {
- Statement wrappedBody = AsyncTransformHelper.wrapWithDeferScope(closure.getCode());
- ClosureExpression newClosure = new ClosureExpression(closure.getParameters(), wrappedBody);
- newClosure.setVariableScope(closure.getVariableScope());
- newClosure.setSourcePosition(closure);
- closure = newClosure;
- }
-
- if (hasYieldReturn) {
- // Inject synthetic $__asyncGen__ as first parameter
- Parameter genParam = AsyncTransformHelper.createGenParam();
- Parameter[] existingParams = closure.getParameters();
- boolean hasUserParams = existingParams != null && existingParams.length > 0;
- Parameter[] newParams;
- if (hasUserParams) {
- newParams = new Parameter[existingParams.length + 1];
- newParams[0] = genParam;
- System.arraycopy(existingParams, 0, newParams, 1, existingParams.length);
- } else {
- newParams = new Parameter[]{genParam};
- }
- ClosureExpression genClosure = new ClosureExpression(newParams, closure.getCode());
- genClosure.setVariableScope(closure.getVariableScope());
- genClosure.setSourcePosition(closure);
- return configureAST(AsyncTransformHelper.buildAsyncGeneratorCall(
- new ArgumentListExpression(genClosure)), ctx);
- } else {
- return configureAST(AsyncTransformHelper.buildAsyncCall(
- new ArgumentListExpression(closure)), ctx);
- }
+ return configureAST(AsyncTransformHelper.transformAsyncClosure(closure), ctx);
}
@Override
diff --git a/src/main/java/org/apache/groovy/runtime/async/AsyncExecutors.java b/src/main/java/org/apache/groovy/runtime/async/AsyncExecutors.java
new file mode 100644
index 0000000000..8114747b17
--- /dev/null
+++ b/src/main/java/org/apache/groovy/runtime/async/AsyncExecutors.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.groovy.runtime.async;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Executor and scheduler configuration for the async runtime.
+ * <p>
+ * Package-private implementation detail of {@link AsyncSupport}. On JDK 21+
+ * the default executor is a virtual-thread-per-task executor; earlier JDKs use a
+ * bounded cached daemon pool sized by {@code groovy.async.parallelism}.
+ *
+ * @since 6.0.0
+ */
+final class AsyncExecutors {
+
+ private static final boolean VIRTUAL_THREADS_AVAILABLE;
+ private static final Executor VIRTUAL_THREAD_EXECUTOR;
+ private static final Executor FALLBACK_EXECUTOR;
+
+ static {
+ Executor vtExecutor = null;
+ boolean vtAvailable = false;
+ try {
+ MethodHandle mh = MethodHandles.lookup().findStatic(
+ Executors.class, "newVirtualThreadPerTaskExecutor",
+ MethodType.methodType(ExecutorService.class));
+ vtExecutor = (Executor) mh.invoke();
+ vtAvailable = true;
+ } catch (Throwable ignored) {
+ // JDK < 21 — virtual threads not available
+ }
+ VIRTUAL_THREAD_EXECUTOR = vtExecutor;
+ VIRTUAL_THREADS_AVAILABLE = vtAvailable;
+
+ int maxThreads = getIntegerSafe("groovy.async.parallelism", 256);
+ if (!VIRTUAL_THREADS_AVAILABLE) {
+ FALLBACK_EXECUTOR = new ThreadPoolExecutor(
+ 0, maxThreads,
+ 60L, TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ @SuppressWarnings("deprecation")
+ long id = t.getId();
+ t.setName("groovy-async-" + id);
+ return t;
+ },
+ new ThreadPoolExecutor.CallerRunsPolicy());
+ } else {
+ FALLBACK_EXECUTOR = null;
+ }
+ }
+
+ private static volatile Executor defaultExecutor = createDefaultExecutor();
+
+ private static final ScheduledExecutorService SCHEDULER =
+ Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "groovy-async-scheduler");
+ t.setDaemon(true);
+ return t;
+ });
+
+ private AsyncExecutors() { }
+
+ static boolean isVirtualThreadsAvailable() {
+ return VIRTUAL_THREADS_AVAILABLE;
+ }
+
+ static Executor getExecutor() {
+ return defaultExecutor;
+ }
+
+ /**
+ * Sets the executor used for async tasks. Passing {@code null} restores the
+ * platform default (virtual threads on JDK 21+, cached pool otherwise).
+ */
+ static void setExecutor(Executor executor) {
+ defaultExecutor = executor != null ? executor : createDefaultExecutor();
+ }
+
+ static void resetExecutor() {
+ defaultExecutor = createDefaultExecutor();
+ }
+
+ static ScheduledExecutorService getScheduler() {
+ return SCHEDULER;
+ }
+
+ private static Executor createDefaultExecutor() {
+ return VIRTUAL_THREADS_AVAILABLE ? VIRTUAL_THREAD_EXECUTOR : FALLBACK_EXECUTOR;
+ }
+
+ private static int getIntegerSafe(String name, int defaultValue) {
+ try {
+ return Integer.getInteger(name, defaultValue);
+ } catch (SecurityException ignore) {
+ return defaultValue;
+ }
+ }
+}
diff --git a/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java b/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java
index 588f5a3084..6c78b0990b 100644
--- a/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java
+++ b/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java
@@ -23,13 +23,9 @@ import groovy.concurrent.Awaitable;
import groovy.concurrent.AwaitableAdapterRegistry;
import java.io.Closeable;
-import java.lang.invoke.MethodHandle;
-import java.lang.invoke.MethodHandles;
-import java.lang.invoke.MethodType;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.ArrayDeque;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Deque;
@@ -43,26 +39,22 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Supplier;
/**
* Internal runtime support for the {@code async}/{@code await}/{@code defer} language features.
* <p>
- * This class contains the actual implementation invoked by compiler-generated code.
- * User code should prefer the static methods on {@link groovy.concurrent.Awaitable}
- * for combinators and configuration.
+ * This class is the entry point invoked by compiler-generated code and also
+ * exposes the combinator and configuration surface used by
+ * {@link groovy.concurrent.Awaitable}. Combinator algorithms live in
+ * {@link AwaitCombinators}; executor/scheduler configuration lives in
+ * {@link AsyncExecutors}.
* <p>
* <b>Thread pool configuration:</b>
* <ul>
@@ -71,7 +63,8 @@ import java.util.function.Supplier;
* <li>On JDK 17-20 the fallback is a cached daemon thread pool
* whose maximum size is controlled by the system property
* {@code groovy.async.parallelism} (default: {@code 256}).</li>
- * <li>The executor can be overridden at any time via {@link #setExecutor}.</li>
+ * <li>The executor can be overridden via {@link #setExecutor}; pass
+ * {@code null} to restore the platform default.</li>
* </ul>
* <p>
* <b>Exception handling</b> follows a transparency principle: the
@@ -82,82 +75,42 @@ import java.util.function.Supplier;
*/
public class AsyncSupport {
- private static final boolean VIRTUAL_THREADS_AVAILABLE;
- private static final Executor VIRTUAL_THREAD_EXECUTOR;
- private static final int FALLBACK_MAX_THREADS;
- private static final Executor FALLBACK_EXECUTOR;
-
- static {
- Executor vtExecutor = null;
- boolean vtAvailable = false;
- try {
- MethodHandle mh = MethodHandles.lookup().findStatic(
- Executors.class, "newVirtualThreadPerTaskExecutor",
- MethodType.methodType(ExecutorService.class));
- vtExecutor = (Executor) mh.invoke();
- vtAvailable = true;
- } catch (Throwable ignored) {
- // JDK < 21 — virtual threads not available
- }
- VIRTUAL_THREAD_EXECUTOR = vtExecutor;
- VIRTUAL_THREADS_AVAILABLE = vtAvailable;
-
- FALLBACK_MAX_THREADS = getIntegerSafe("groovy.async.parallelism", 256);
- if (!VIRTUAL_THREADS_AVAILABLE) {
- FALLBACK_EXECUTOR = new ThreadPoolExecutor(
- 0, FALLBACK_MAX_THREADS,
- 60L, TimeUnit.SECONDS,
- new SynchronousQueue<>(),
- r -> {
- Thread t = new Thread(r);
- t.setDaemon(true);
- @SuppressWarnings("deprecation")
- long id = t.getId();
- t.setName("groovy-async-" + id);
- return t;
- },
- new ThreadPoolExecutor.CallerRunsPolicy());
- } else {
- FALLBACK_EXECUTOR = null;
- }
- }
-
- private static volatile Executor defaultExecutor = createDefaultExecutor();
-
- private static final ScheduledExecutorService SCHEDULER =
- Executors.newSingleThreadScheduledExecutor(r -> {
- Thread t = new Thread(r, "groovy-async-scheduler");
- t.setDaemon(true);
- return t;
- });
-
private AsyncSupport() { }
// ---- executor configuration -----------------------------------------
/** Returns the shared scheduler for delays, timeouts, and scope deadlines. */
public static ScheduledExecutorService getScheduler() {
- return SCHEDULER;
+ return AsyncExecutors.getScheduler();
}
/** Returns {@code true} if running on JDK 21+ with virtual thread support. */
public static boolean isVirtualThreadsAvailable() {
- return VIRTUAL_THREADS_AVAILABLE;
+ return AsyncExecutors.isVirtualThreadsAvailable();
}
/** Returns the current executor used for async tasks. */
public static Executor getExecutor() {
- return defaultExecutor;
+ return AsyncExecutors.getExecutor();
}
- /** Sets the executor used for async tasks. */
+ /**
+ * Sets the executor used for async tasks.
+ * <p>
+ * Pass {@code null} to restore the platform default (virtual threads on
+ * JDK 21+, cached daemon pool otherwise). The change takes effect
+ * for subsequent {@code async} launches; in-flight tasks keep the
+ * executor that started them.
+ *
+ * @param executor the executor to use, or {@code null} to reset
+ */
public static void setExecutor(Executor executor) {
- defaultExecutor = Objects.requireNonNull(executor, "executor must not be null");
+ AsyncExecutors.setExecutor(executor);
}
- /** Resets the executor to the default (virtual threads on JDK 21+, cached pool otherwise). */
+ /** Resets the executor to the platform default. Equivalent to {@code setExecutor(null)}. */
public static void resetExecutor() {
- defaultExecutor = createDefaultExecutor();
+ AsyncExecutors.resetExecutor();
}
// ---- await overloads ------------------------------------------------
@@ -211,7 +164,12 @@ public class AsyncSupport {
/**
* Awaits an arbitrary object by adapting it via {@link Awaitable#from(Object)}.
- * This is the fallback overload called by compiler-generated await expressions.
+ * <p>
+ * This is the fallback overload called by compiler-generated {@code await}
+ * expressions. The compiler inserts a cast to {@link Object} so that
+ * overload resolution does not have to choose among {@code Awaitable},
+ * {@code CompletionStage}, and {@code Future} when a value implements more
+ * than one of them (e.g. {@link CompletableFuture}).
*/
@SuppressWarnings("unchecked")
public static <T> T await(Object source) {
@@ -231,7 +189,7 @@ public class AsyncSupport {
*/
public static <T> Awaitable<T> executeAsync(Supplier<T> supplier, Executor executor) {
Objects.requireNonNull(supplier, "supplier must not be null");
- Executor targetExecutor = executor != null ? executor : defaultExecutor;
+ Executor targetExecutor = executor != null ? executor : AsyncExecutors.getExecutor();
return GroovyPromise.of(CompletableFuture.supplyAsync(() -> {
try {
return supplier.get();
@@ -245,11 +203,12 @@ public class AsyncSupport {
* Executes the given supplier asynchronously using the default executor.
*/
public static <T> Awaitable<T> async(Supplier<T> supplier) {
- return executeAsync(supplier, defaultExecutor);
+ return executeAsync(supplier, AsyncExecutors.getExecutor());
}
/**
- * Lightweight task spawn. Executes the supplier asynchronously using the default executor.
+ * Lightweight task spawn. Alias of {@link #async(Supplier)} for Go-style
+ * ergonomics; semantics are identical.
*/
public static <T> Awaitable<T> go(Supplier<T> supplier) {
return async(supplier);
@@ -286,7 +245,6 @@ public class AsyncSupport {
* subsequent exceptions are added as suppressed. If a deferred action returns
* a Future/Awaitable, the result is awaited before continuing.
*/
- @SuppressWarnings("unchecked")
public static void executeDeferScope(Deque<Callable<?>> scope) {
if (scope == null || scope.isEmpty()) return;
Throwable firstError = null;
@@ -349,11 +307,10 @@ public class AsyncSupport {
* @param <T> the element type
* @return an Iterable that yields values from the generator
*/
- @SuppressWarnings("unchecked")
public static <T> Iterable<T> asyncGenerator(Consumer<Object> body) {
Objects.requireNonNull(body, "body must not be null");
GeneratorBridge<T> bridge = new GeneratorBridge<>();
- defaultExecutor.execute(() -> {
+ AsyncExecutors.getExecutor().execute(() -> {
try {
body.accept(bridge);
bridge.complete();
@@ -387,14 +344,15 @@ public class AsyncSupport {
return () -> iter;
}
if (source instanceof Object[]) return (Iterable<T>) Arrays.asList((Object[]) source);
- // Try adapter registry
return AwaitableAdapterRegistry.toIterable(source);
}
/**
* Closes a source if it implements {@link Closeable} or
* {@link AutoCloseable}. Called by compiler-generated finally block
- * in {@code for await} loops.
+ * in {@code for await} loops. Cleanup exceptions are swallowed so they
+ * cannot mask the original loop error; prefer robust {@code close()}
+ * implementations.
*/
public static void closeIterable(Object source) {
if (source instanceof Closeable c) {
@@ -410,31 +368,31 @@ public class AsyncSupport {
}
}
- // ---- combinators ----------------------------------------------------
+ // ---- combinators (delegate to AwaitCombinators) ---------------------
/**
* Waits for all given sources to complete, returning their results in order.
- * Multi-arg {@code await(a, b, c)} desugars to this.
+ * Multi-arg {@code await(a, b, c)} desugars to the non-blocking
+ * {@link #allAsync(Object...)} form, then awaits it.
*/
public static <T> List<T> all(Object... sources) {
- CompletableFuture<?>[] futures = toCombinatorFutures(sources);
- CompletableFuture.allOf(futures).join();
- return getJoinedResults(futures);
+ return AwaitCombinators.all(sources);
}
/**
* Returns the result of the first source to complete (success or failure).
*/
public static <T> T any(Object... sources) {
- return AsyncSupport.await(AsyncSupport.<T>anyAsync(sources).toCompletableFuture());
+ return AwaitCombinators.any(sources);
}
/**
* Returns the result of the first source to complete <em>successfully</em>.
- * Only fails when all sources fail.
+ * Only fails when all sources fail (aggregate {@link CompletionException};
+ * {@code await} transparency rethrows the cause).
*/
public static <T> T first(Object... sources) {
- return AsyncSupport.await(AsyncSupport.<T>firstAsync(sources).toCompletableFuture());
+ return AwaitCombinators.first(sources);
}
/**
@@ -442,136 +400,27 @@ public class AsyncSupport {
* {@link AwaitResult} without throwing.
*/
public static List<AwaitResult<Object>> allSettled(Object... sources) {
- return AsyncSupport.await(allSettledAsync(sources).toCompletableFuture());
+ return AwaitCombinators.allSettled(sources);
}
- // ---- async combinator variants (return Awaitable, non-blocking) ------
-
/** Non-blocking variant of {@link #all} — returns an Awaitable. */
public static Awaitable<List<Object>> allAsync(Object... sources) {
- CompletableFuture<?>[] futures = toCombinatorFutures(sources);
-
- // allOf is naturally fail-fast: it completes as soon as any
- // source fails OR all sources succeed. We track the first
- // failure explicitly because allOf doesn't guarantee which
- // exception propagates when multiple futures fail.
- var firstError = new AtomicReference<Throwable>();
- for (CompletableFuture<?> f : futures) {
- f.whenComplete((v, e) -> {
- if (e != null) firstError.compareAndSet(null, e);
- });
- }
-
- CompletableFuture<List<Object>> combined = CompletableFuture.allOf(futures)
- .thenApply(v -> getJoinedResults(futures));
-
- // Replace allOf's arbitrary exception with the temporally-first one
- CompletableFuture<List<Object>> withFirstError = combined.exceptionally(e -> {
- Throwable first = firstError.get();
- if (first != null && first != e && first != e.getCause()) {
- throw first instanceof CompletionException ce ? ce : new CompletionException(first);
- }
- throw e instanceof CompletionException ce ? ce : new CompletionException(e);
- });
- return GroovyPromise.of(withFirstError);
+ return AwaitCombinators.allAsync(sources);
}
/** Non-blocking variant of {@link #any} — returns an Awaitable. */
- @SuppressWarnings("unchecked")
public static <T> Awaitable<T> anyAsync(Object... sources) {
- return (Awaitable<T>) GroovyPromise.of(CompletableFuture.anyOf(toCombinatorFutures(sources)));
+ return AwaitCombinators.anyAsync(sources);
}
/** Non-blocking variant of {@link #first} — returns an Awaitable. */
- @SuppressWarnings("unchecked")
public static <T> Awaitable<T> firstAsync(Object... sources) {
- CompletableFuture<T>[] futures = toFirstCombinatorFutures(sources);
- CompletableFuture<T> result = new CompletableFuture<>();
- var remainingFailures = new AtomicInteger(futures.length);
- List<Throwable> errors = Collections.synchronizedList(new ArrayList<>(futures.length));
- for (CompletableFuture<T> future : futures) {
- future.whenComplete((value, error) -> {
- if (error == null) {
- result.complete(value);
- return;
- }
- errors.add(error);
- if (remainingFailures.decrementAndGet() == 0) {
- result.completeExceptionally(aggregateFirstFailures(futures.length, errors));
- }
- });
- }
- return GroovyPromise.of(result);
+ return AwaitCombinators.firstAsync(sources);
}
/** Non-blocking variant of {@link #allSettled} — returns an Awaitable. */
public static Awaitable<List<AwaitResult<Object>>> allSettledAsync(Object... sources) {
- CompletableFuture<?>[] futures = toCombinatorFutures(sources);
- CompletableFuture<List<AwaitResult<Object>>> combined = CompletableFuture.allOf(
- Arrays.stream(futures)
- .map(future -> future.handle((value, error) -> null))
- .toArray(CompletableFuture[]::new)
- )
- .thenApply(v -> getAwaitResults(futures));
- return GroovyPromise.of(combined);
- }
-
- @SuppressWarnings("unchecked")
- private static <T> List<T> getJoinedResults(CompletableFuture<?>[] futures) {
- List<T> results = new ArrayList<>(futures.length);
- for (CompletableFuture<?> future : futures) {
- results.add((T) future.join());
- }
- return results;
- }
-
- private static List<AwaitResult<Object>> getAwaitResults(CompletableFuture<?>[] futures) {
- List<AwaitResult<Object>> results = new ArrayList<>(futures.length);
- for (CompletableFuture<?> f : futures) {
- try {
- results.add(AwaitResult.success(f.join()));
- } catch (CompletionException e) {
- results.add(AwaitResult.failure(unwrap(e)));
- } catch (CancellationException e) {
- results.add(AwaitResult.failure(e));
- }
- }
- return results;
- }
-
- private static CompletableFuture<?>[] toCombinatorFutures(Object... sources) {
- return Arrays.stream(sources)
- .map(source -> Awaitable.from(source).toCompletableFuture())
- .toArray(CompletableFuture[]::new);
- }
-
- @SuppressWarnings("unchecked")
- private static <T> CompletableFuture<T>[] toFirstCombinatorFutures(Object... sources) {
- validateFirstSources(sources);
- return (CompletableFuture<T>[]) toCombinatorFutures(sources);
- }
-
- private static void validateFirstSources(Object[] sources) {
- if (sources == null) {
- throw new IllegalArgumentException("sources must not be null");
- }
- if (sources.length == 0) {
- throw new IllegalArgumentException("sources must not be empty");
- }
- for (Object source : sources) {
- if (source == null) {
- throw new IllegalArgumentException("sources must not contain null elements");
- }
- }
- }
-
- private static CompletionException aggregateFirstFailures(int sourceCount, List<Throwable> errors) {
- CompletionException aggregate = new CompletionException(
- "All " + sourceCount + " tasks failed", errors.get(0));
- for (int i = 1; i < errors.size(); i++) {
- aggregate.addSuppressed(errors.get(i));
- }
- return aggregate;
+ return AwaitCombinators.allSettledAsync(sources);
}
// ---- delay and timeout ----------------------------------------------
@@ -580,27 +429,26 @@ public class AsyncSupport {
* Returns an Awaitable that completes after the specified delay.
*/
public static Awaitable<Void> delay(long millis) {
- CompletableFuture<Void> future = new CompletableFuture<>();
- SCHEDULER.schedule(() -> future.complete(null), millis, TimeUnit.MILLISECONDS);
- return GroovyPromise.of(future);
+ return delay(millis, TimeUnit.MILLISECONDS);
}
/** Delay with explicit time unit. */
public static Awaitable<Void> delay(long duration, TimeUnit unit) {
CompletableFuture<Void> future = new CompletableFuture<>();
- SCHEDULER.schedule(() -> future.complete(null), duration, unit);
+ AsyncExecutors.getScheduler().schedule(() -> future.complete(null), duration, unit);
return GroovyPromise.of(future);
}
/**
* Wraps a source with a timeout. If the source does not complete within
- * the specified time, the returned Awaitable fails with {@link TimeoutException}.
+ * the specified time, the returned Awaitable fails with {@link TimeoutException}
+ * and the underlying computation is cancelled.
*/
@SuppressWarnings("unchecked")
public static <T> Awaitable<T> orTimeout(Object source, long timeout, TimeUnit unit) {
CompletableFuture<T> future = (CompletableFuture<T>) Awaitable.from(source).toCompletableFuture();
CompletableFuture<T> result = new CompletableFuture<>();
- ScheduledFuture<?> timer = SCHEDULER.schedule(() -> {
+ ScheduledFuture<?> timer = AsyncExecutors.getScheduler().schedule(() -> {
if (!result.isDone()) {
result.completeExceptionally(new TimeoutException("Timed out after " + timeout + " " + unit));
future.cancel(true);
@@ -621,12 +469,13 @@ public class AsyncSupport {
/**
* Wraps a source with a timeout that uses a fallback value instead of throwing.
+ * On timeout the underlying computation is cancelled.
*/
@SuppressWarnings("unchecked")
public static <T> Awaitable<T> completeOnTimeout(Object source, T fallback, long timeout, TimeUnit unit) {
CompletableFuture<T> future = (CompletableFuture<T>) Awaitable.from(source).toCompletableFuture();
CompletableFuture<T> result = new CompletableFuture<>();
- ScheduledFuture<?> timer = SCHEDULER.schedule(() -> {
+ ScheduledFuture<?> timer = AsyncExecutors.getScheduler().schedule(() -> {
if (!result.isDone()) {
result.complete(fallback);
future.cancel(true);
@@ -658,6 +507,11 @@ public class AsyncSupport {
return null; // unreachable
}
+ /**
+ * Strips JDK wrapper layers ({@link CompletionException},
+ * {@link ExecutionException}, {@link InvocationTargetException},
+ * {@link UndeclaredThrowableException}) to expose the original cause.
+ */
public static Throwable unwrap(Throwable t) {
while ((t instanceof CompletionException || t instanceof ExecutionException
|| t instanceof InvocationTargetException
@@ -675,22 +529,10 @@ public class AsyncSupport {
// ---- internal utilities ---------------------------------------------
- private static Executor createDefaultExecutor() {
- return VIRTUAL_THREADS_AVAILABLE ? VIRTUAL_THREAD_EXECUTOR : FALLBACK_EXECUTOR;
- }
-
private static CancellationException interruptedAwait(String message, InterruptedException cause) {
Thread.currentThread().interrupt();
CancellationException cancellation = new CancellationException(message);
cancellation.initCause(cause);
return cancellation;
}
-
- private static int getIntegerSafe(String name, int defaultValue) {
- try {
- return Integer.getInteger(name, defaultValue);
- } catch (SecurityException ignore) {
- return defaultValue;
- }
- }
}
diff --git a/src/main/java/org/apache/groovy/runtime/async/AwaitCombinators.java b/src/main/java/org/apache/groovy/runtime/async/AwaitCombinators.java
new file mode 100644
index 0000000000..9e103818a8
--- /dev/null
+++ b/src/main/java/org/apache/groovy/runtime/async/AwaitCombinators.java
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.groovy.runtime.async;
+
+import groovy.concurrent.AwaitResult;
+import groovy.concurrent.Awaitable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Combinators for joining multiple async sources ({@code all}, {@code any},
+ * {@code first}, {@code allSettled}).
+ * <p>
+ * Package-private implementation detail of {@link AsyncSupport}. Blocking
+ * variants are thin wrappers over the non-blocking ones via
+ * {@link AsyncSupport#await(Object)}, so each combinator has a single
+ * implementation of its joining algorithm.
+ *
+ * @since 6.0.0
+ */
+final class AwaitCombinators {
+
+ private AwaitCombinators() { }
+
+ // ---- blocking -------------------------------------------------------
+
+ @SuppressWarnings("unchecked")
+ static <T> List<T> all(Object... sources) {
+ return (List<T>) AsyncSupport.await(allAsync(sources));
+ }
+
+ static <T> T any(Object... sources) {
+ return AsyncSupport.await(AwaitCombinators.<T>anyAsync(sources));
+ }
+
+ static <T> T first(Object... sources) {
+ return AsyncSupport.await(AwaitCombinators.<T>firstAsync(sources));
+ }
+
+ static List<AwaitResult<Object>> allSettled(Object... sources) {
+ return AsyncSupport.await(allSettledAsync(sources));
+ }
+
+ // ---- non-blocking ---------------------------------------------------
+
+ static Awaitable<List<Object>> allAsync(Object... sources) {
+ CompletableFuture<?>[] futures = toFutures(sources);
+
+ // allOf is fail-fast but does not guarantee which exception propagates
+ // when several futures fail; track the temporally-first error ourselves.
+ var firstError = new AtomicReference<Throwable>();
+ for (CompletableFuture<?> future : futures) {
+ future.whenComplete((value, error) -> {
+ if (error != null) {
+ firstError.compareAndSet(null, error);
+ }
+ });
+ }
+
+ CompletableFuture<List<Object>> combined = CompletableFuture.allOf(futures)
+ .thenApply(ignored -> getJoinedResults(futures));
+
+ CompletableFuture<List<Object>> withFirstError = combined.exceptionally(error -> {
+ Throwable first = firstError.get();
+ if (first != null && first != error && first != error.getCause()) {
+ throw asCompletionException(first);
+ }
+ throw asCompletionException(error);
+ });
+ return GroovyPromise.of(withFirstError);
+ }
+
+ @SuppressWarnings("unchecked")
+ static <T> Awaitable<T> anyAsync(Object... sources) {
+ return (Awaitable<T>) GroovyPromise.of(CompletableFuture.anyOf(toFutures(sources)));
+ }
+
+ @SuppressWarnings("unchecked")
+ static <T> Awaitable<T> firstAsync(Object... sources) {
+ validateNonEmptySources(sources);
+ CompletableFuture<T>[] futures = (CompletableFuture<T>[]) toFutures(sources);
+ CompletableFuture<T> result = new CompletableFuture<>();
+ var remainingFailures = new AtomicInteger(futures.length);
+ List<Throwable> errors = Collections.synchronizedList(new ArrayList<>(futures.length));
+ for (CompletableFuture<T> future : futures) {
+ future.whenComplete((value, error) -> {
+ if (error == null) {
+ result.complete(value);
+ return;
+ }
+ errors.add(error);
+ if (remainingFailures.decrementAndGet() == 0) {
+ result.completeExceptionally(aggregateFirstFailures(futures.length, errors));
+ }
+ });
+ }
+ return GroovyPromise.of(result);
+ }
+
+ static Awaitable<List<AwaitResult<Object>>> allSettledAsync(Object... sources) {
+ CompletableFuture<?>[] futures = toFutures(sources);
+ CompletableFuture<List<AwaitResult<Object>>> combined = CompletableFuture.allOf(
+ Arrays.stream(futures)
+ .map(future -> future.handle((value, error) -> null))
+ .toArray(CompletableFuture[]::new))
+ .thenApply(ignored -> getAwaitResults(futures));
+ return GroovyPromise.of(combined);
+ }
+
+ // ---- helpers --------------------------------------------------------
+
+ private static CompletableFuture<?>[] toFutures(Object... sources) {
+ if (sources == null) {
+ return new CompletableFuture<?>[0];
+ }
+ CompletableFuture<?>[] futures = new CompletableFuture<?>[sources.length];
+ for (int i = 0; i < sources.length; i++) {
+ futures[i] = Awaitable.from(sources[i]).toCompletableFuture();
+ }
+ return futures;
+ }
+
+ private static void validateNonEmptySources(Object[] sources) {
+ if (sources == null) {
+ throw new IllegalArgumentException("sources must not be null");
+ }
+ if (sources.length == 0) {
+ throw new IllegalArgumentException("sources must not be empty");
+ }
+ for (Object source : sources) {
+ if (source == null) {
+ throw new IllegalArgumentException("sources must not contain null elements");
+ }
+ }
+ }
+
+ /**
+ * Builds the aggregate {@link CompletionException} for {@code first}
+ * when every source fails. The first failure is the cause; remaining
+ * failures are attached as suppressed exceptions.
+ */
+ private static CompletionException aggregateFirstFailures(int sourceCount, List<Throwable> errors) {
+ CompletionException aggregate = new CompletionException(
+ "All " + sourceCount + " tasks failed", errors.get(0));
+ for (int i = 1; i < errors.size(); i++) {
+ aggregate.addSuppressed(errors.get(i));
+ }
+ return aggregate;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> List<T> getJoinedResults(CompletableFuture<?>[] futures) {
+ List<T> results = new ArrayList<>(futures.length);
+ for (CompletableFuture<?> future : futures) {
+ results.add((T) future.join());
+ }
+ return results;
+ }
+
+ private static List<AwaitResult<Object>> getAwaitResults(CompletableFuture<?>[] futures) {
+ List<AwaitResult<Object>> results = new ArrayList<>(futures.length);
+ for (CompletableFuture<?> future : futures) {
+ try {
+ results.add(AwaitResult.success(future.join()));
+ } catch (CompletionException e) {
+ results.add(AwaitResult.failure(AsyncSupport.unwrap(e)));
+ } catch (CancellationException e) {
+ results.add(AwaitResult.failure(e));
+ }
+ }
+ return results;
+ }
+
+ private static CompletionException asCompletionException(Throwable error) {
+ return error instanceof CompletionException ce ? ce : new CompletionException(error);
+ }
+}
diff --git a/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java b/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java
index 044b948d0f..fa8bdf6803 100644
--- a/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java
+++ b/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java
@@ -18,7 +18,6 @@
*/
package org.apache.groovy.runtime.async;
-// AsyncContext support reserved for future enhancement
import groovy.concurrent.Awaitable;
import java.util.Objects;
@@ -179,9 +178,7 @@ public class GroovyPromise<T> implements Awaitable<T> {
* {@inheritDoc}
* <p>
* Returns a new {@code GroovyPromise} whose result is obtained by applying
- * the given function to this promise's result. The current
- * {@code AsyncContext} snapshot is captured when the continuation is
- * registered and restored when it executes.
+ * the given function to this promise's result when it completes.
*/
@Override
public <U> Awaitable<U> then(Function<? super T, ? extends U> fn) {
@@ -193,8 +190,6 @@ public class GroovyPromise<T> implements Awaitable<T> {
* <p>
* Returns a new {@code GroovyPromise} that is the result of composing this
* promise with the async function, enabling flat-mapping of awaitables.
- * The current {@code AsyncContext} snapshot is captured when the
- * continuation is registered and restored when it executes.
*/
@Override
public <U> Awaitable<U> thenCompose(Function<? super T, ? extends Awaitable<U>> fn) {
@@ -208,8 +203,6 @@ public class GroovyPromise<T> implements Awaitable<T> {
* Returns a new {@code GroovyPromise} that handles exceptions thrown by this promise.
* The throwable passed to the handler is deeply unwrapped to strip JDK
* wrapper layers ({@code CompletionException}, {@code ExecutionException}).
- * The handler runs with the {@code AsyncContext} snapshot that was active
- * when the recovery continuation was registered.
*/
@Override
public Awaitable<T> exceptionally(Function<Throwable, ? extends T> fn) {
diff --git a/src/main/java/org/apache/groovy/runtime/async/package-info.java b/src/main/java/org/apache/groovy/runtime/async/package-info.java
index f40eece142..5b03865bd0 100644
--- a/src/main/java/org/apache/groovy/runtime/async/package-info.java
+++ b/src/main/java/org/apache/groovy/runtime/async/package-info.java
@@ -18,6 +18,24 @@
*/
/**
- * Runtime support for async/await expressions. Provides continuation-passing style transformation and execution framework.
+ * Runtime support for Groovy's {@code async}/{@code await}/{@code defer}
+ * language features.
+ * <p>
+ * Layout:
+ * <ul>
+ * <li>{@link org.apache.groovy.runtime.async.AsyncSupport} — public entry
+ * point used by compiler-generated code and by
+ * {@link groovy.concurrent.Awaitable}</li>
+ * <li>{@code AsyncExecutors} — executor/scheduler configuration</li>
+ * <li>{@code AwaitCombinators} — {@code all}/{@code any}/{@code first}/
+ * {@code allSettled} joining algorithms</li>
+ * <li>{@link org.apache.groovy.runtime.async.GroovyPromise} — default
+ * {@link groovy.concurrent.Awaitable} implementation</li>
+ * <li>{@link org.apache.groovy.runtime.async.GeneratorBridge} —
+ * producer/consumer bridge for {@code yield return}</li>
+ * <li>{@link org.apache.groovy.runtime.async.DefaultAsyncScope},
+ * {@link org.apache.groovy.runtime.async.DefaultAsyncChannel} —
+ * structured concurrency and CSP channel implementations</li>
+ * </ul>
*/
package org.apache.groovy.runtime.async;
diff --git a/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java b/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java
index f4275e6246..41f96d184d 100644
--- a/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java
+++ b/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java
@@ -22,20 +22,22 @@ import groovy.concurrent.Awaitable;
import org.apache.groovy.runtime.async.AsyncSupport;
import org.codehaus.groovy.ast.ClassHelper;
import org.codehaus.groovy.ast.ClassNode;
-import org.codehaus.groovy.ast.CodeVisitorSupport;
import org.codehaus.groovy.ast.Parameter;
import org.codehaus.groovy.ast.expr.ArgumentListExpression;
import org.codehaus.groovy.ast.expr.ClosureExpression;
import org.codehaus.groovy.ast.expr.Expression;
import org.codehaus.groovy.ast.expr.StaticMethodCallExpression;
import org.codehaus.groovy.ast.query.AstQuery;
+import org.codehaus.groovy.ast.stmt.ForStatement;
import org.codehaus.groovy.ast.stmt.Statement;
+import org.codehaus.groovy.ast.stmt.TryCatchStatement;
+
+import java.util.concurrent.atomic.AtomicLong;
import static org.codehaus.groovy.ast.tools.GeneralUtils.args;
import static org.codehaus.groovy.ast.tools.GeneralUtils.block;
import static org.codehaus.groovy.ast.tools.GeneralUtils.callX;
import static org.codehaus.groovy.ast.tools.GeneralUtils.castX;
-import static org.codehaus.groovy.ast.tools.GeneralUtils.classX;
import static org.codehaus.groovy.ast.tools.GeneralUtils.declS;
import static org.codehaus.groovy.ast.tools.GeneralUtils.stmt;
import static org.codehaus.groovy.ast.tools.GeneralUtils.tryCatchS;
@@ -44,7 +46,9 @@ import static org.codehaus.groovy.ast.tools.GeneralUtils.varX;
/**
* Shared AST utilities for the {@code async}/{@code await}/{@code defer} language features.
* <p>
- * Centralises AST node construction for the parser ({@code AstBuilder}).
+ * Centralises AST node construction and higher-level rewrites used by the
+ * parser ({@code AstBuilder}) so feature logic does not sprawl across the
+ * visitor.
*
* @since 6.0.0
*/
@@ -65,6 +69,9 @@ public final class AsyncTransformHelper {
private static final String DEFER_METHOD = "defer";
private static final String EXECUTE_DEFER_SCOPE_METHOD = "executeDeferScope";
+ /** Monotonic suffix for synthetic {@code for await} source variables. */
+ private static final AtomicLong FOR_AWAIT_SEQ = new AtomicLong();
+
private AsyncTransformHelper() { }
private static Expression ensureArgs(Expression expr) {
@@ -74,17 +81,25 @@ public final class AsyncTransformHelper {
/**
* Builds {@code AsyncSupport.await(arg)} or for multi-arg:
* {@code AsyncSupport.await(Awaitable.all(arg1, arg2, ...))}.
+ * <p>
+ * Single-arg form casts to {@link Object} so static/dynamic overload
+ * selection targets {@link AsyncSupport#await(Object)}, avoiding
+ * ambiguity when the argument implements multiple async interfaces
+ * (e.g. {@link java.util.concurrent.CompletableFuture} implements both
+ * {@link java.util.concurrent.CompletionStage} and
+ * {@link java.util.concurrent.Future}).
*/
public static Expression buildAwaitCall(Expression arg) {
if (arg instanceof ArgumentListExpression args && args.getExpressions().size() > 1) {
- Expression allCall = callX(classX(AWAITABLE_TYPE), "all", args);
+ Expression allCall = callX(AWAITABLE_TYPE, "all", args);
return callX(ASYNC_SUPPORT_TYPE, AWAIT_METHOD, new ArgumentListExpression(allCall));
}
- // Cast to Object to force dynamic dispatch to the await(Object) overload,
- // avoiding ambiguity when the argument implements multiple async interfaces
- // (e.g., CompletableFuture implements both CompletionStage and Future)
+ Expression source = arg;
+ if (arg instanceof ArgumentListExpression args && !args.getExpressions().isEmpty()) {
+ source = args.getExpression(0);
+ }
return callX(ASYNC_SUPPORT_TYPE, AWAIT_METHOD,
- new ArgumentListExpression(castX(ClassHelper.OBJECT_TYPE, arg)));
+ new ArgumentListExpression(castX(ClassHelper.OBJECT_TYPE, source)));
}
/**
@@ -143,34 +158,12 @@ public final class AsyncTransformHelper {
public static boolean containsYieldReturn(Statement stmt) {
return AstQuery.from(stmt)
.descendants(StaticMethodCallExpression.class)
- .notInto(ClosureExpression.class) // don't descend into nested closures
+ .notInto(ClosureExpression.class)
.where(call -> YIELD_RETURN_METHOD.equals(call.getMethod())
&& AsyncSupport.class.getName().equals(call.getOwnerType().getName()))
.any();
}
- /**
- * Rewrites {@code yieldReturn(expr)} calls to {@code yieldReturn($__asyncGen__, expr)}
- * by injecting the generator parameter reference.
- */
- public static void injectGenParamIntoYieldReturnCalls(Statement stmt, Parameter genParam) {
- stmt.visit(new CodeVisitorSupport() {
- @Override
- public void visitStaticMethodCallExpression(StaticMethodCallExpression call) {
- if (YIELD_RETURN_METHOD.equals(call.getMethod())
- && AsyncSupport.class.getName().equals(call.getOwnerType().getName())) {
- // Already built with gen param by buildYieldReturnCall — no action needed
- // This method is a hook point for future transformations
- }
- super.visitStaticMethodCallExpression(call);
- }
- @Override
- public void visitClosureExpression(ClosureExpression expression) {
- // Don't descend into nested closures
- }
- });
- }
-
/**
* Returns {@code true} if the statement tree contains a {@code defer} call,
* without descending into nested closures.
@@ -178,7 +171,7 @@ public final class AsyncTransformHelper {
public static boolean containsDefer(Statement stmt) {
return AstQuery.from(stmt)
.descendants(StaticMethodCallExpression.class)
- .notInto(ClosureExpression.class) // don't descend into nested closures
+ .notInto(ClosureExpression.class)
.where(call -> DEFER_METHOD.equals(call.getMethod())
&& AsyncSupport.class.getName().equals(call.getOwnerType().getName()))
.any();
@@ -193,14 +186,80 @@ public final class AsyncTransformHelper {
* </pre>
*/
public static Statement wrapWithDeferScope(Statement body) {
- // var $__deferScope__ = AsyncSupport.createDeferScope()
Statement declStmt = declS(varX(DEFER_SCOPE_VAR),
callX(ASYNC_SUPPORT_TYPE, CREATE_DEFER_SCOPE_METHOD));
- // try { body } finally { AsyncSupport.executeDeferScope($__deferScope__) }
Statement finallyStmt = stmt(callX(ASYNC_SUPPORT_TYPE, EXECUTE_DEFER_SCOPE_METHOD,
args(varX(DEFER_SCOPE_VAR))));
return block(declStmt, tryCatchS(body, finallyStmt));
}
+
+ /**
+ * Rewrites a {@code for await} loop into a block that materialises the
+ * source iterable, runs the loop, and closes the source in a finally block:
+ * <pre>
+ * var $__forAwaitSource_N = AsyncSupport.toIterable(collection)
+ * try {
+ * for (... in $__forAwaitSource_N) { body }
+ * } finally {
+ * AsyncSupport.closeIterable($__forAwaitSource_N)
+ * }
+ * </pre>
+ *
+ * @param forStatement the for statement whose collection is the await source
+ * @return the rewritten statement block
+ */
+ public static Statement wrapForAwaitLoop(ForStatement forStatement) {
+ Expression original = forStatement.getCollectionExpression();
+ String tempVar = "$__forAwaitSource_" + FOR_AWAIT_SEQ.incrementAndGet();
+ Expression toIterableCall = buildToIterableCall(original);
+
+ Statement declStmt = declS(varX(tempVar), toIterableCall);
+ forStatement.setCollectionExpression(varX(tempVar));
+
+ Statement finallyStmt = stmt(buildCloseIterableCall(varX(tempVar)));
+ TryCatchStatement tryCatch = new TryCatchStatement(forStatement, finallyStmt);
+ return block(declStmt, tryCatch);
+ }
+
+ /**
+ * Applies defer-scope wrapping and generator parameter injection to an
+ * {@code async { ... }} closure, then builds the runtime call
+ * ({@code async} or {@code asyncGenerator}).
+ *
+ * @param closure the parsed async closure expression
+ * @return the desugared runtime call expression
+ */
+ public static Expression transformAsyncClosure(ClosureExpression closure) {
+ boolean hasYieldReturn = containsYieldReturn(closure.getCode());
+ boolean hasDefer = containsDefer(closure.getCode());
+
+ if (hasDefer) {
+ Statement wrappedBody = wrapWithDeferScope(closure.getCode());
+ ClosureExpression newClosure = new ClosureExpression(closure.getParameters(), wrappedBody);
+ newClosure.setVariableScope(closure.getVariableScope());
+ newClosure.setSourcePosition(closure);
+ closure = newClosure;
+ }
+
+ if (hasYieldReturn) {
+ Parameter genParam = createGenParam();
+ Parameter[] existingParams = closure.getParameters();
+ boolean hasUserParams = existingParams != null && existingParams.length > 0;
+ Parameter[] newParams;
+ if (hasUserParams) {
+ newParams = new Parameter[existingParams.length + 1];
+ newParams[0] = genParam;
+ System.arraycopy(existingParams, 0, newParams, 1, existingParams.length);
+ } else {
+ newParams = new Parameter[]{genParam};
+ }
+ ClosureExpression genClosure = new ClosureExpression(newParams, closure.getCode());
+ genClosure.setVariableScope(closure.getVariableScope());
+ genClosure.setSourcePosition(closure);
+ return buildAsyncGeneratorCall(new ArgumentListExpression(genClosure));
+ }
+ return buildAsyncCall(new ArgumentListExpression(closure));
+ }
}
diff --git a/src/spec/doc/core-async-await.adoc b/src/spec/doc/core-async-await.adoc
index 858a52bcda..dfff725c31 100644
--- a/src/spec/doc/core-async-await.adoc
+++ b/src/spec/doc/core-async-await.adoc
@@ -359,7 +359,9 @@ import org.apache.groovy.runtime.async.AsyncSupport
import java.util.concurrent.Executors
AsyncSupport.setExecutor(Executors.newFixedThreadPool(4))
-AsyncSupport.resetExecutor() // restore default
+AsyncSupport.setExecutor(null) // restore platform default
+// or equivalently:
+AsyncSupport.resetExecutor()
----
[[async-jdk-integration]]
diff --git a/src/test/groovy/groovy/AsyncAwaitTest.groovy b/src/test/groovy/groovy/AsyncAwaitTest.groovy
index ffc5826a34..92085273c8 100644
--- a/src/test/groovy/groovy/AsyncAwaitTest.groovy
+++ b/src/test/groovy/groovy/AsyncAwaitTest.groovy
@@ -1116,11 +1116,51 @@ final class AsyncAwaitTest {
await Awaitable.first(f1, f2)
assert false : 'should have thrown'
} catch (Exception e) {
+ // await exception transparency peels the aggregate CompletionException
assert e != null
}
'''
}
+ @Test
+ void testSetExecutorNullResetsToDefault() {
+ assertScript '''
+ import groovy.concurrent.Awaitable
+ import org.apache.groovy.runtime.async.AsyncSupport
+ import java.util.concurrent.Executors
+
+ def original = Awaitable.getExecutor()
+ def custom = Executors.newSingleThreadExecutor()
+ try {
+ Awaitable.setExecutor(custom)
+ assert Awaitable.getExecutor().is(custom)
+ Awaitable.setExecutor(null)
+ assert Awaitable.getExecutor() != null
+ assert !Awaitable.getExecutor().is(custom)
+ // AsyncSupport path is equivalent
+ AsyncSupport.setExecutor(custom)
+ AsyncSupport.setExecutor(null)
+ assert AsyncSupport.getExecutor() != null
+ } finally {
+ Awaitable.setExecutor(original)
+ custom.shutdown()
+ }
+ '''
+ }
+
+ @Test
+ void testAwaitObjectFallbackHandlesNullAndAwaitable() {
+ assertScript '''
+ import org.apache.groovy.runtime.async.AsyncSupport
+ import groovy.concurrent.Awaitable
+
+ assert AsyncSupport.await((Object) null) == null
+ assert AsyncSupport.await((Object) Awaitable.of(42)) == 42
+ def task = async { 'x' }
+ assert AsyncSupport.await((Object) task) == 'x'
+ '''
+ }
+
@Test
void testAllSettledWithCancelledTask() {
assertScript '''
diff --git a/src/test/groovy/groovy/concurrent/AwaitResultTest.groovy b/src/test/groovy/groovy/concurrent/AwaitResultTest.groovy
new file mode 100644
index 0000000000..72da8c98e6
--- /dev/null
+++ b/src/test/groovy/groovy/concurrent/AwaitResultTest.groovy
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent
+
+import org.junit.jupiter.api.Test
+
+import static groovy.test.GroovyAssert.shouldFail
+
+/**
+ * Unit tests for {@link AwaitResult} value-object behaviour.
+ */
+final class AwaitResultTest {
+
+ @Test
+ void testSuccessHoldsValueIncludingNull() {
+ def ok = AwaitResult.success('hello')
+ assert ok.success
+ assert !ok.failure
+ assert ok.value == 'hello'
+ assert ok.toString() == 'AwaitResult.Success[hello]'
+
+ def nil = AwaitResult.success(null)
+ assert nil.success
+ assert nil.value == null
+ }
+
+ @Test
+ void testFailureHoldsError() {
+ def err = new RuntimeException('boom')
+ def fail = AwaitResult.failure(err)
+ assert fail.failure
+ assert !fail.success
+ assert fail.error.is(err)
+ assert fail.toString().contains('boom')
+ }
+
+ @Test
+ void testFailureRejectsNullError() {
+ shouldFail(NullPointerException) {
+ AwaitResult.failure(null)
+ }
+ }
+
+ @Test
+ void testGetValueOnFailureThrows() {
+ def fail = AwaitResult.failure(new Exception('x'))
+ def e = shouldFail(IllegalStateException) {
+ fail.value
+ }
+ assert e.message.contains('failed')
+ }
+
+ @Test
+ void testGetErrorOnSuccessThrows() {
+ def ok = AwaitResult.success(1)
+ def e = shouldFail(IllegalStateException) {
+ ok.error
+ }
+ assert e.message.contains('successful')
+ }
+
+ @Test
+ void testGetOrElse() {
+ assert AwaitResult.success(7).getOrElse { -1 } == 7
+ assert AwaitResult.failure(new Exception('e')).getOrElse { it.message.length() } == 1
+ }
+
+ @Test
+ void testMapOnSuccessAndFailure() {
+ assert AwaitResult.success('ab').map { it.length() }.value == 2
+ def fail = AwaitResult.<String>failure(new Exception('keep'))
+ def mapped = fail.map { it.toUpperCase() }
+ assert mapped.failure
+ assert mapped.error.message == 'keep'
+ }
+
+ @Test
+ void testMapRejectsNullFunction() {
+ shouldFail(NullPointerException) {
+ AwaitResult.success(1).map(null)
+ }
+ }
+
+ @Test
+ void testEqualsAndHashCode() {
+ def a = AwaitResult.success('x')
+ def b = AwaitResult.success('x')
+ def c = AwaitResult.success('y')
+ def err = new RuntimeException('e')
+ def f1 = AwaitResult.failure(err)
+ def f2 = AwaitResult.failure(err)
+
+ assert a == b
+ assert a.hashCode() == b.hashCode()
+ assert a != c
+ assert a != f1
+ assert f1 == f2
+ assert f1.hashCode() == f2.hashCode()
+ assert a != null
+ assert a != 'x'
+ }
+}
diff --git a/src/test/groovy/org/apache/groovy/runtime/async/AsyncSupportInternalsTest.groovy b/src/test/groovy/org/apache/groovy/runtime/async/AsyncSupportInternalsTest.groovy
new file mode 100644
index 0000000000..ce237007e5
--- /dev/null
+++ b/src/test/groovy/org/apache/groovy/runtime/async/AsyncSupportInternalsTest.groovy
@@ -0,0 +1,259 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.groovy.runtime.async
+
+import groovy.concurrent.Awaitable
+import org.junit.jupiter.api.Test
+
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionException
+import java.util.concurrent.CompletionStage
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
+
+import static groovy.test.GroovyAssert.shouldFail
+
+/**
+ * Focused unit coverage for the async runtime refactor
+ * ({@link AsyncSupport}, {@link AsyncExecutors}, {@link AwaitCombinators}).
+ */
+final class AsyncSupportInternalsTest {
+
+ @Test
+ void testGoIsAliasOfAsync() {
+ def viaGo = AsyncSupport.go { 11 }
+ def viaAsync = AsyncSupport.async { 22 }
+ assert AsyncSupport.await(viaGo) == 11
+ assert AsyncSupport.await(viaAsync) == 22
+ }
+
+ @Test
+ void testAwaitObjectOnCompletableFutureImplementsMultipleInterfaces() {
+ // CompletableFuture is CompletionStage and Future; await(Object) is unambiguous
+ CompletableFuture<String> cf = CompletableFuture.completedFuture('cf')
+ assert AsyncSupport.await((Object) cf) == 'cf'
+ assert AsyncSupport.await((CompletionStage<String>) cf) == 'cf'
+ assert AsyncSupport.await(cf) == 'cf'
+ }
+
+ @Test
+ void testAwaitUnwrapsNestedWrappers() {
+ def cause = new IllegalStateException('root')
+ def cf = new CompletableFuture()
+ cf.completeExceptionally(new CompletionException(new ExecutionException(cause)))
+ def thrown = shouldFail(IllegalStateException) {
+ AsyncSupport.await(cf)
+ }
+ assert thrown.is(cause)
+ }
+
+ @Test
+ void testAllAsyncEmptyAndSingle() {
+ assert AsyncSupport.await(AsyncSupport.allAsync()) == []
+ assert AsyncSupport.await(AsyncSupport.allAsync(Awaitable.of(1))) == [1]
+ assert AsyncSupport.all() == []
+ assert AsyncSupport.all(Awaitable.of('a'), Awaitable.of('b')) == ['a', 'b']
+ }
+
+ @Test
+ void testAllPropagatesOriginalFailureNotWrapper() {
+ def fail = Awaitable.failed(new RuntimeException('all-x'))
+ def ok = Awaitable.of(1)
+ def thrown = shouldFail(RuntimeException) {
+ AsyncSupport.all(ok, fail)
+ }
+ assert thrown.message == 'all-x'
+ }
+
+ @Test
+ void testFirstAllFailThrowsUnwrappedCause() {
+ def f1 = Awaitable.failed(new RuntimeException('e1'))
+ def f2 = Awaitable.failed(new RuntimeException('e2'))
+ // Aggregate is a CompletionException; await transparency peels to the cause
+ def thrown = shouldFail(RuntimeException) {
+ AsyncSupport.await(AsyncSupport.firstAsync(f1, f2))
+ }
+ assert thrown.message == 'e1'
+ }
+
+ @Test
+ void testFirstRejectsNullArrayAndNullElementAndEmpty() {
+ shouldFail(IllegalArgumentException) {
+ AsyncSupport.firstAsync((Object[]) null)
+ }
+ shouldFail(IllegalArgumentException) {
+ AsyncSupport.firstAsync()
+ }
+ shouldFail(IllegalArgumentException) {
+ AsyncSupport.firstAsync(Awaitable.of(1), null)
+ }
+ }
+
+ @Test
+ void testAnyAsyncPicksCompletedValue() {
+ def result = AsyncSupport.await(AsyncSupport.anyAsync(
+ Awaitable.of('winner'),
+ asyncSleep(500, 'loser')))
+ assert result == 'winner'
+ }
+
+ @Test
+ void testAllSettledMixed() {
+ def results = AsyncSupport.allSettled(
+ Awaitable.of('ok'),
+ Awaitable.failed(new RuntimeException('bad')))
+ assert results.size() == 2
+ assert results[0].success && results[0].value == 'ok'
+ assert results[1].failure && results[1].error.message == 'bad'
+ }
+
+ @Test
+ void testSetExecutorNullAndReset() {
+ def original = AsyncSupport.getExecutor()
+ def custom = Executors.newSingleThreadExecutor()
+ try {
+ AsyncSupport.setExecutor(custom)
+ assert AsyncSupport.getExecutor().is(custom)
+ AsyncSupport.setExecutor(null)
+ assert AsyncSupport.getExecutor() != null
+ assert !AsyncSupport.getExecutor().is(custom)
+ AsyncSupport.setExecutor(custom)
+ AsyncSupport.resetExecutor()
+ assert !AsyncSupport.getExecutor().is(custom)
+ } finally {
+ AsyncSupport.setExecutor(original)
+ custom.shutdownNow()
+ }
+ }
+
+ @Test
+ void testDelayAndTimeouts() {
+ assert AsyncSupport.await(AsyncSupport.delay(1)) == null
+ assert AsyncSupport.await(AsyncSupport.delay(1, TimeUnit.MILLISECONDS)) == null
+
+ def slow = asyncSleep(5_000, 'late')
+ def timedOut = shouldFail(TimeoutException) {
+ AsyncSupport.await(AsyncSupport.orTimeoutMillis(slow, 20))
+ }
+ assert timedOut.message.contains('Timed out')
+
+ def fallback = AsyncSupport.await(
+ AsyncSupport.completeOnTimeoutMillis(asyncSleep(5_000, 'late'), 'fb', 20))
+ assert fallback == 'fb'
+ }
+
+ @Test
+ void testToIterableAndCloseIterable() {
+ assert AsyncSupport.toIterable(null).iterator().toList() == []
+ assert AsyncSupport.toIterable([1, 2]).iterator().toList() == [1, 2]
+ assert AsyncSupport.toIterable((Object[]) ['a', 'b']).iterator().toList() == ['a', 'b']
+ def it = [9].iterator()
+ assert AsyncSupport.toIterable(it).iterator().is(it)
+
+ def closed = new boolean[1]
+ def closeable = new Closeable() {
+ @Override void close() { closed[0] = true }
+ }
+ AsyncSupport.closeIterable(closeable)
+ assert closed[0]
+
+ def autoClosed = new boolean[1]
+ def auto = new AutoCloseable() {
+ @Override void close() { autoClosed[0] = true }
+ }
+ AsyncSupport.closeIterable(auto)
+ assert autoClosed[0]
+
+ // non-closeable is a no-op
+ AsyncSupport.closeIterable('plain')
+ }
+
+ @Test
+ void testCloseIterableSwallowsCloseException() {
+ def bad = new Closeable() {
+ @Override void close() { throw new IOException('ignore-me') }
+ }
+ AsyncSupport.closeIterable(bad) // must not throw
+ }
+
+ @Test
+ void testDeferScopeLIFOAndNullGuards() {
+ def log = []
+ def scope = AsyncSupport.createDeferScope()
+ AsyncSupport.defer(scope, { log << 'a'; null })
+ AsyncSupport.defer(scope, { log << 'b'; null })
+ AsyncSupport.executeDeferScope(scope)
+ assert log == ['b', 'a']
+
+ shouldFail(IllegalStateException) {
+ AsyncSupport.defer(null, { null })
+ }
+ shouldFail(IllegalArgumentException) {
+ AsyncSupport.defer(AsyncSupport.createDeferScope(), null)
+ }
+ AsyncSupport.executeDeferScope(null) // no-op
+ AsyncSupport.executeDeferScope(AsyncSupport.createDeferScope()) // empty no-op
+ }
+
+ @Test
+ void testYieldReturnRequiresGeneratorBridge() {
+ shouldFail(IllegalStateException) {
+ AsyncSupport.yieldReturn('not-a-bridge', 1)
+ }
+ }
+
+ @Test
+ void testAsyncGeneratorProducesValues() {
+ def values = AsyncSupport.<Integer>asyncGenerator { bridge ->
+ AsyncSupport.yieldReturn(bridge, 1)
+ AsyncSupport.yieldReturn(bridge, 2)
+ }.iterator().toList()
+ assert values == [1, 2]
+ }
+
+ @Test
+ void testWrapForFutureAndUnwrap() {
+ def ce = new CompletionException(new RuntimeException('x'))
+ assert AsyncSupport.wrapForFuture(ce).is(ce)
+ def wrapped = AsyncSupport.wrapForFuture(new RuntimeException('y'))
+ assert wrapped instanceof CompletionException
+ assert AsyncSupport.unwrap(wrapped).message == 'y'
+ }
+
+ @Test
+ void testIsVirtualThreadsAvailableIsConsistent() {
+ boolean available = AsyncSupport.isVirtualThreadsAvailable()
+ assert available == Awaitable.isVirtualThreadsAvailable()
+ }
+
+ @Test
+ void testSchedulerIsDaemon() {
+ def scheduler = AsyncSupport.getScheduler()
+ assert scheduler != null
+ }
+
+ private static Awaitable asyncSleep(long millis, Object value) {
+ AsyncSupport.async {
+ Thread.sleep(millis)
+ value
+ }
+ }
+}
diff --git a/src/test/groovy/org/codehaus/groovy/transform/AsyncTransformHelperTest.groovy b/src/test/groovy/org/codehaus/groovy/transform/AsyncTransformHelperTest.groovy
new file mode 100644
index 0000000000..129591aede
--- /dev/null
+++ b/src/test/groovy/org/codehaus/groovy/transform/AsyncTransformHelperTest.groovy
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.codehaus.groovy.transform
+
+import org.codehaus.groovy.ast.ClassHelper
+import org.codehaus.groovy.ast.Parameter
+import org.codehaus.groovy.ast.expr.ArgumentListExpression
+import org.codehaus.groovy.ast.expr.CastExpression
+import org.codehaus.groovy.ast.expr.ClosureExpression
+import org.codehaus.groovy.ast.expr.ConstantExpression
+import org.codehaus.groovy.ast.expr.MethodCallExpression
+import org.codehaus.groovy.ast.expr.StaticMethodCallExpression
+import org.codehaus.groovy.ast.expr.VariableExpression
+import org.codehaus.groovy.ast.stmt.BlockStatement
+import org.codehaus.groovy.ast.stmt.EmptyStatement
+import org.codehaus.groovy.ast.stmt.ExpressionStatement
+import org.codehaus.groovy.ast.stmt.ForStatement
+import org.codehaus.groovy.ast.tools.GeneralUtils
+import org.junit.jupiter.api.Test
+
+/**
+ * Unit tests for {@link AsyncTransformHelper} AST construction helpers.
+ */
+final class AsyncTransformHelperTest {
+
+ @Test
+ void testBuildAwaitCallSingleArgUsesAwaitWithObjectCast() {
+ def expr = AsyncTransformHelper.buildAwaitCall(new ConstantExpression(1))
+ assert methodName(expr) == 'await'
+ assert ownerName(expr).endsWith('AsyncSupport')
+ def arg0 = ((StaticMethodCallExpression) expr).arguments.expressions[0]
+ assert arg0 instanceof CastExpression
+ assert arg0.type == ClassHelper.OBJECT_TYPE
+ }
+
+ @Test
+ void testBuildAwaitCallMultiArgUsesAwaitableAllThenAwait() {
+ def args = new ArgumentListExpression(
+ new ConstantExpression(1),
+ new ConstantExpression(2))
+ def expr = AsyncTransformHelper.buildAwaitCall(args)
+ assert methodName(expr) == 'await'
+ def arg0 = ((StaticMethodCallExpression) expr).arguments.expressions[0]
+ assert methodName(arg0) == 'all'
+ assert ownerName(arg0).endsWith('Awaitable')
+ }
+
+ @Test
+ void testContainsYieldReturnAndDeferDoNotDescendIntoNestedClosures() {
+ def yieldCall = AsyncTransformHelper.buildYieldReturnCall(new ConstantExpression(1))
+ def deferCall = AsyncTransformHelper.buildDeferCall(
+ new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE))
+
+ def outerWithYield = new BlockStatement()
+ outerWithYield.addStatement(new ExpressionStatement(yieldCall))
+ assert AsyncTransformHelper.containsYieldReturn(outerWithYield)
+ assert !AsyncTransformHelper.containsDefer(outerWithYield)
+
+ def outerWithDefer = new BlockStatement()
+ outerWithDefer.addStatement(new ExpressionStatement(deferCall))
+ assert AsyncTransformHelper.containsDefer(outerWithDefer)
+ assert !AsyncTransformHelper.containsYieldReturn(outerWithDefer)
+
+ // Nested inside a child closure — must be ignored
+ def nested = new ClosureExpression(Parameter.EMPTY_ARRAY, outerWithYield)
+ def outerOnlyNested = new BlockStatement()
+ outerOnlyNested.addStatement(new ExpressionStatement(nested))
+ assert !AsyncTransformHelper.containsYieldReturn(outerOnlyNested)
+ }
+
+ @Test
+ void testWrapWithDeferScopeStructure() {
+ def body = EmptyStatement.INSTANCE
+ def wrapped = AsyncTransformHelper.wrapWithDeferScope(body)
+ assert wrapped instanceof BlockStatement
+ assert wrapped.statements.size() == 2
+ }
+
+ @Test
+ void testWrapForAwaitLoopUsesUniqueSyntheticNames() {
+ def collection = new ConstantExpression([1, 2, 3])
+ def loop1 = new ForStatement(
+ new Parameter(org.codehaus.groovy.ast.ClassHelper.OBJECT_TYPE, 'item'),
+ collection,
+ EmptyStatement.INSTANCE)
+ def loop2 = new ForStatement(
+ new Parameter(org.codehaus.groovy.ast.ClassHelper.OBJECT_TYPE, 'item'),
+ collection,
+ EmptyStatement.INSTANCE)
+
+ def block1 = AsyncTransformHelper.wrapForAwaitLoop(loop1) as BlockStatement
+ def block2 = AsyncTransformHelper.wrapForAwaitLoop(loop2) as BlockStatement
+
+ def name1 = extractDeclaredName(block1)
+ def name2 = extractDeclaredName(block2)
+ assert name1.startsWith('$__forAwaitSource_')
+ assert name2.startsWith('$__forAwaitSource_')
+ assert name1 != name2
+ }
+
+ @Test
+ void testTransformAsyncClosurePlainAndGenerator() {
+ def plain = new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE)
+ def plainCall = AsyncTransformHelper.transformAsyncClosure(plain)
+ assert methodName(plainCall) == 'async'
+
+ def yieldCall = AsyncTransformHelper.buildYieldReturnCall(new ConstantExpression(1))
+ def yieldBody = new BlockStatement()
+ yieldBody.addStatement(new ExpressionStatement(yieldCall))
+ def genClosure = new ClosureExpression(Parameter.EMPTY_ARRAY, yieldBody)
+ def genCall = AsyncTransformHelper.transformAsyncClosure(genClosure)
+ assert methodName(genCall) == 'asyncGenerator'
+ }
+
+ @Test
+ void testTransformAsyncClosureWithDefer() {
+ def deferCall = AsyncTransformHelper.buildDeferCall(
+ new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE))
+ def body = new BlockStatement()
+ body.addStatement(new ExpressionStatement(deferCall))
+ def closure = new ClosureExpression(Parameter.EMPTY_ARRAY, body)
+ def call = AsyncTransformHelper.transformAsyncClosure(closure)
+ assert methodName(call) == 'async'
+ }
+
+ @Test
+ void testCreateGenParamAndBuilders() {
+ def p = AsyncTransformHelper.createGenParam()
+ assert p.name == '$__asyncGen__'
+
+ assert methodName(AsyncTransformHelper.buildAsyncCall(
+ new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE))) == 'async'
+ assert methodName(AsyncTransformHelper.buildAsyncGeneratorCall(
+ new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE))) == 'asyncGenerator'
+ assert methodName(AsyncTransformHelper.buildToIterableCall(new ConstantExpression([]))) == 'toIterable'
+ assert methodName(AsyncTransformHelper.buildCloseIterableCall(GeneralUtils.varX('x'))) == 'closeIterable'
+ }
+
+ private static String methodName(Object expr) {
+ if (expr instanceof StaticMethodCallExpression) {
+ return expr.method
+ }
+ if (expr instanceof MethodCallExpression) {
+ return expr.methodAsString
+ }
+ throw new AssertionError("unexpected expression: ${expr?.getClass()?.name}")
+ }
+
+ private static String ownerName(Object expr) {
+ if (expr instanceof StaticMethodCallExpression) {
+ return expr.ownerType.name
+ }
+ if (expr instanceof MethodCallExpression) {
+ return expr.objectExpression.type.name
+ }
+ throw new AssertionError("unexpected expression: ${expr?.getClass()?.name}")
+ }
+
+ private static String extractDeclaredName(BlockStatement block) {
+ def declStmt = block.statements[0] as ExpressionStatement
+ def decl = declStmt.expression
+ def left = decl.leftExpression
+ if (left instanceof VariableExpression) {
+ return left.name
+ }
+ return decl.variableExpression.name
+ }
+}