GROOVY-12181: Refactor async runtime and AST helpers introduced by GROOVY-9381

src/main/java/groovy/concurrent/AwaitResult.java

@@ -69,9 +69,8 @@ private AwaitResult(T value, Throwable error, boolean success) {
      * @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);
     }
 
     /**

src/main/java/groovy/concurrent/Awaitable.java

@@ -439,9 +439,10 @@ static <T> Awaitable<T> any(Object... sources) {
      * 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

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;

src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java

@@ -539,20 +539,7 @@
         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;
@@ -924,7 +911,7 @@
     public Statement visitLabeledStmtAlt(final LabeledStmtAltContext ctx) {
         Statement statement = (Statement) this.visit(ctx.statement());
 
         statement.addStatementLabel(this.visitIdentifier(ctx.identifier()));
 
         return statement;
     }
@@ -964,7 +951,7 @@
     public ExpressionStatement visitDeferStmtAlt(final DeferStmtAltContext ctx) {
         Expression action;
         ExpressionStatement stmtExprStmt = (ExpressionStatement) this.visit(ctx.statementExpression());
         Expression expr = stmtExprStmt.getExpression();
         if (expr instanceof ClosureExpression) {
             action = expr;
         } else {
@@ -3090,39 +3077,7 @@
     @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
@@ -3354,7 +3309,7 @@
                 new BinaryExpression(
                         this.visitVariableNames(ctx.left),
                         this.createGroovyToken(ctx.op),
                         ((ExpressionStatement) this.visit(ctx.right)).getExpression()),
                 ctx);
     }
 
@@ -4660,7 +4615,7 @@
             return (ConstantExpression) expression;
         }
 
         return configureAST(new ConstantExpression(expression.getText()), expression);
     }
 
     private BinaryExpression createBinaryExpression(final ExpressionContext left, final Token op, final ExpressionContext right) {

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&nbsp;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&nbsp;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;
+        }
+    }
+}