Copilot commented on code in PR #2386: URL: https://github.com/apache/groovy/pull/2386#discussion_r2868656297
########## src/test/groovy/org/codehaus/groovy/transform/AsyncVirtualThreadTest.groovy: ########## @@ -0,0 +1,1138 @@ +/* + * 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.junit.jupiter.api.Test + +import static groovy.test.GroovyAssert.assertScript + +/** + * Tests for virtual thread integration, executor configuration, exception + * handling consistency across async methods/closures/lambdas, and coverage + * improvements for the async/await feature. + * + * @since 6.0.0 + */ +final class AsyncVirtualThreadTest { + + // ---- Virtual thread detection and usage ---- + + @Test + void testVirtualThreadsAvailable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + // JDK 21+ should have virtual threads + def jdkVersion = Runtime.version().feature() + if (jdkVersion >= 21) { + assert isVirtualThreadsAvailable() + } + ''' + } + + @Test + void testAsyncMethodRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class VTService { + async checkThread() { + return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def service = new VTService() + def result = service.checkThread().get() + assert result == true + } + ''' + } + + @Test + void testAsyncClosureRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncVirtual = async { Thread.currentThread().isVirtual() } + def awaitable = asyncVirtual() + assert await(awaitable) == true + } + ''' + } + + @Test + void testAsyncLambdaRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncFn = async { x -> Thread.currentThread().isVirtual() } + def result = await(asyncFn(42)) + assert result == true + } + ''' + } + + @Test + void testForAwaitRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.AsyncStream + + class VTGenerator { + async generate() { + yield return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def gen = new VTGenerator() + def stream = gen.generate() + def results = [] + for await (item in stream) { + results << item + } + assert results == [true] + } + ''' + } + + @Test + void testHighConcurrencyWithVirtualThreads() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class HighConcurrency { + async compute(int n) { + Thread.sleep(10) + return n * 2 + } + } + + if (isVirtualThreadsAvailable()) { + def svc = new HighConcurrency() + // Launch 1000 concurrent tasks — trivial with virtual threads + def awaitables = (1..1000).collect { svc.compute(it) } + def results = awaitAll(awaitables as Object[]) + assert results.size() == 1000 + assert results[0] == 2 + assert results[999] == 2000 + } + ''' + } + + // ---- Executor configuration ---- + + @Test + void testCustomExecutorOverride() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.atomic.AtomicReference + + def savedExecutor = getExecutor() + try { + def customPool = Executors.newFixedThreadPool(2, { r -> + def t = new Thread(r) + t.setName("custom-async-" + t.getId()) + t + }) + setExecutor(customPool) + + def asyncName = async { + Thread.currentThread().getName() + } + def awaitable = asyncName() + def threadName = await(awaitable) + assert threadName.startsWith("custom-async-") + } finally { + setExecutor(savedExecutor) + } + ''' + } + + @Test + void testSetExecutorNullResetsToDefault() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + + def originalExecutor = getExecutor() + // Set a custom executor + setExecutor(Executors.newSingleThreadExecutor()) + assert getExecutor() != originalExecutor + // Reset to null — should restore default + setExecutor(null) + def restored = getExecutor() + assert restored != null + // Verify it works + def awaitable = groovy.concurrent.AsyncUtils.async { 42 } + assert groovy.concurrent.AsyncUtils.await(awaitable) == 42 + // Restore original + setExecutor(originalExecutor) + ''' + } + + @Test + void testAsyncMethodWithCustomExecutorField() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executor + import java.util.concurrent.Executors + + class CustomExecutorService { + static Executor myPool = Executors.newFixedThreadPool(1, { r -> + def t = new Thread(r) + t.setName("my-pool-thread") + t + }) + Review Comment: `CustomExecutorService.myPool` is a static fixed thread pool with non-daemon threads and is never shut down. This can leak threads across the full test suite and prevent the JVM from exiting. Consider using daemon threads for the factory and adding explicit shutdown/cleanup (e.g., in a `cleanupSpec`/`@AfterEach` or via `close()` in the script). ########## src/test/groovy/org/codehaus/groovy/transform/AsyncPatternsTest.groovy: ########## @@ -0,0 +1,1006 @@ +/* + * 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 groovy.concurrent.AsyncUtils +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Test + +import static groovy.test.GroovyAssert.assertScript + +/** + * Tests covering async/await patterns including CPU-bound work, I/O-bound + * simulation, caller perspectives, continuations, parallel execution, + * delay, exception handling, async closures/lambdas, and async method + * control flow. + * + * <p>Await usage patterns: + * <ul> + * <li>Inside {@code @Async} methods: {@code await expr} (keyword syntax)</li> + * <li>Inside {@code async { }} closures: {@code await(expr)} with static import</li> + * <li>At script top-level: {@code AsyncUtils.await(expr)} or static import</li> + * </ul> + */ +class AsyncPatternsTest { + + @AfterEach + void resetExecutor() { + AsyncUtils.setExecutor(null) + } + + // ========================================================================= + // CPU-Bound async work + // ========================================================================= + + @Test + void testCpuBoundWithAsyncClosure() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def calculateComplexOutput(String input) { + new StringBuilder(input).reverse().toString() + } + + def calculateResultAsync(String input) { + def task = async { + def compute = async { calculateComplexOutput(input) } + def result = await(compute()) + return result + } + return task() + } + + assert await(calculateResultAsync("Hello, World!")) == "!dlroW ,olleH" + ''' + } + + @Test + void testCpuBoundWithAsyncAnnotation() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class ComputeService { + private String calculateComplexOutput(String input) { + new StringBuilder(input).reverse().toString() + } + + @Async + def calculateResult(String input) { + return calculateComplexOutput(input) + } + } + + def service = new ComputeService() + assert await(service.calculateResult("Groovy")) == "yvoorG" + ''' + } + + // ========================================================================= + // I/O-Bound async simulation + // ========================================================================= + + @Test + void testIoBoundSimulated() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync(String url) { + def fetch = async { + await(delay(50)) + return "Response from ${url}" + } + return fetch() + } + + def downloadDataAsync() { + def download = async { + def result = await(fetchDataAsync("https://example.com/data";)) + return result + } + return download() + } + + assert await(downloadDataAsync()) == "Response from https://example.com/data"; + ''' + } + + // ========================================================================= + // Caller perspectives + // ========================================================================= + + @Test + void testCallerAwaitImmediately() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { + await(delay(50)) + return "fetched data" + } + return fetch() + } + + assert await(fetchDataAsync()) == "fetched data" + ''' + } + + @Test + void testCallerStartThenAwaitLater() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { + await(delay(50)) + return "fetched data" + } + return fetch() + } + + def caller = async { + def task = fetchDataAsync() + def localResult = 1 + 2 + 3 + def downloadResult = await(task) + return "${downloadResult} (local: ${localResult})" + } + def result = await(caller()) + assert result == "fetched data (local: 6)" + ''' + } + + @Test + void testCallerReturnsTaskWithoutAwaiting() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { + await(delay(30)) + return "data" + } + return fetch() + } + + def caller() { + def task = fetchDataAsync() + return task + } + + assert await(caller()) == "data" + ''' + } + + // ========================================================================= + // Task continuations + // ========================================================================= + + @Test + void testTaskContinuationWithThen() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { "Hello, World!" } + return fetch() + } + + def calculateComplexOutput(String input) { + new StringBuilder(input).reverse().toString() + } + + def result = await( + fetchDataAsync().then { data -> calculateComplexOutput(data) } + ) + assert result == "!dlroW ,olleH" + ''' + } + + @Test + void testTaskContinuationChaining() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def chain = async { 10 } + def result = await( + chain() + .then { it * 2 } + .then { it + 5 } + .then { "Result: $it" } + ) + assert result == "Result: 25" + ''' + } + + @Test + void testTaskContinuationWithThenCompose() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchUser(int id) { + def user = async { [id: id, name: "User${id}"] } + return user() + } + + def fetchOrders(int userId) { + def orders = async { ["order1", "order2"] } + return orders() + } + + def result = await( + fetchUser(42).thenCompose { user -> + fetchOrders(user.id).then { orders -> + [user: user, orders: orders] + } + } + ) + + assert result.user.name == "User42" + assert result.orders == ["order1", "order2"] + ''' + } + + // ========================================================================= + // Parallel execution: awaitAll / awaitAny + // ========================================================================= + + @Test + void testAwaitAllParallelTasks() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchPage(String url) { + def page = async { + await(delay(30)) + return "Content of ${url} (${url.length()} chars)" + } + return page() + } + + def results = awaitAll( + fetchPage("https://www.google.com/";), + fetchPage("https://www.microsoft.com/";), + fetchPage("https://www.example.com/";) + ) + + assert results.size() == 3 + assert results[0].contains("google") + assert results[1].contains("microsoft") + assert results[2].contains("example") + ''' + } + + @Test + void testAwaitAnyRace() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fastTask = async { + await(delay(10)) + return "fast" + } + def fast = fastTask() + def slowTask = async { + await(delay(2000)) Review Comment: `awaitAny(fast, slow)` returns as soon as `fast` completes, but `slowTask` will continue waiting for 2 seconds afterwards, consuming an executor thread/virtual thread and slowing the suite. Consider reducing the delay significantly and/or cancelling the losing task after `awaitAny` returns to keep tests resource-friendly. ```suggestion await(delay(20)) ``` ########## src/test/groovy/org/codehaus/groovy/transform/AsyncVirtualThreadTest.groovy: ########## @@ -0,0 +1,1138 @@ +/* + * 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.junit.jupiter.api.Test + +import static groovy.test.GroovyAssert.assertScript + +/** + * Tests for virtual thread integration, executor configuration, exception + * handling consistency across async methods/closures/lambdas, and coverage + * improvements for the async/await feature. + * + * @since 6.0.0 + */ +final class AsyncVirtualThreadTest { + + // ---- Virtual thread detection and usage ---- + + @Test + void testVirtualThreadsAvailable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + // JDK 21+ should have virtual threads + def jdkVersion = Runtime.version().feature() + if (jdkVersion >= 21) { + assert isVirtualThreadsAvailable() + } + ''' + } + + @Test + void testAsyncMethodRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class VTService { + async checkThread() { + return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def service = new VTService() + def result = service.checkThread().get() + assert result == true + } + ''' + } + + @Test + void testAsyncClosureRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncVirtual = async { Thread.currentThread().isVirtual() } + def awaitable = asyncVirtual() + assert await(awaitable) == true + } + ''' + } + + @Test + void testAsyncLambdaRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncFn = async { x -> Thread.currentThread().isVirtual() } + def result = await(asyncFn(42)) + assert result == true + } + ''' + } + + @Test + void testForAwaitRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.AsyncStream + + class VTGenerator { + async generate() { + yield return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def gen = new VTGenerator() + def stream = gen.generate() + def results = [] + for await (item in stream) { + results << item + } + assert results == [true] + } + ''' + } + + @Test + void testHighConcurrencyWithVirtualThreads() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class HighConcurrency { + async compute(int n) { + Thread.sleep(10) + return n * 2 + } + } + + if (isVirtualThreadsAvailable()) { + def svc = new HighConcurrency() + // Launch 1000 concurrent tasks — trivial with virtual threads + def awaitables = (1..1000).collect { svc.compute(it) } + def results = awaitAll(awaitables as Object[]) + assert results.size() == 1000 + assert results[0] == 2 + assert results[999] == 2000 + } + ''' + } + + // ---- Executor configuration ---- + + @Test + void testCustomExecutorOverride() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.atomic.AtomicReference + + def savedExecutor = getExecutor() + try { + def customPool = Executors.newFixedThreadPool(2, { r -> + def t = new Thread(r) + t.setName("custom-async-" + t.getId()) + t + }) + setExecutor(customPool) Review Comment: This test creates a fixed thread pool (`customPool`) with non-daemon threads and never shuts it down. Even though `setExecutor(savedExecutor)` is called, the pool will keep threads alive and can hang the test JVM. Please ensure the pool is shut down (preferably in the `finally`) and/or use daemon threads for the thread factory. ########## src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java: ########## @@ -0,0 +1,144 @@ +/* + * 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 java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Function; + +/** + * Default {@link Awaitable} implementation backed by a {@link CompletableFuture}. + * <p> + * This is the concrete type returned by {@code async} methods. It delegates + * all operations to an underlying {@code CompletableFuture} while keeping the + * public API limited to the {@code Awaitable} contract, thereby decoupling + * user code from JDK-specific async APIs. + * <p> + * This class is an internal implementation detail and should not be referenced + * directly by user code. Use the {@link Awaitable} interface instead. + * + * @param <T> the result type + * @see Awaitable + * @since 6.0.0 + */ +public class GroovyPromise<T> implements Awaitable<T> { + + private final CompletableFuture<T> future; + + public GroovyPromise(CompletableFuture<T> future) { + this.future = Objects.requireNonNull(future, "future must not be null"); + } + + /** + * Creates a {@code GroovyPromise} wrapping the given {@link CompletableFuture}. + */ + public static <T> GroovyPromise<T> of(CompletableFuture<T> future) { + return new GroovyPromise<>(future); + } + + @Override + public T get() throws InterruptedException, ExecutionException { + return future.get(); + } + + @Override + public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + return future.get(timeout, unit); + } + + @Override + public boolean isDone() { + return future.isDone(); + } + + /** + * Attempts to cancel this computation. Delegates to + * {@link CompletableFuture#cancel(boolean) CompletableFuture.cancel(true)}. + * <p> + * <b>Note:</b> {@code CompletableFuture} cancellation sets the future's state + * to cancelled but does <em>not</em> reliably interrupt the underlying thread. + * Async work already in progress may continue running in the background. + * For cooperative cancellation, check {@link Thread#isInterrupted()} in + * long-running async bodies. + * + * @return {@code true} if the future was successfully cancelled + */ + @Override + public boolean cancel() { + return future.cancel(true); + } + + @Override + public boolean isCancelled() { + return future.isCancelled(); + } + + @Override + public boolean isCompletedExceptionally() { + return future.isCompletedExceptionally(); + } + + @Override + public <U> Awaitable<U> then(Function<? super T, ? extends U> fn) { + return new GroovyPromise<>(future.thenApply(fn)); + } + + @Override + public <U> Awaitable<U> thenCompose(Function<? super T, ? extends Awaitable<U>> fn) { + return new GroovyPromise<>(future.thenCompose(t -> fn.apply(t).toCompletableFuture())); + } + + @Override + public Awaitable<T> exceptionally(Function<Throwable, ? extends T> fn) { + return new GroovyPromise<>(future.exceptionally(t -> { + // Unwrap all wrapper layers so handler sees the original exception + Throwable cause = t; + while (cause.getCause() != null + && (cause instanceof CompletionException + || cause instanceof ExecutionException + || cause instanceof java.lang.reflect.UndeclaredThrowableException + || cause instanceof java.lang.reflect.InvocationTargetException)) { + cause = cause.getCause(); + } + return fn.apply(cause); + })); Review Comment: `GroovyPromise.exceptionally` reimplements deep-unwrapping logic that also exists in `AsyncSupport.deepUnwrap()`. Duplicating the wrapper list in multiple places risks the two implementations drifting. Consider delegating to a single shared utility (e.g., `AsyncSupport.deepUnwrap`) so exception unwrapping behavior stays consistent across `await()` and `exceptionally()`. ########## src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java: ########## @@ -465,14 +467,60 @@ public Statement visitLoopStmtAlt(final LoopStmtAltContext ctx) { } @Override - public ForStatement visitForStmtAlt(final ForStmtAltContext ctx) { + public Statement visitForStmtAlt(final ForStmtAltContext ctx) { + // 'for await' async iteration + if (ctx.AWAIT() != null) { + return visitForAwait(ctx); + } + Function<Statement, ForStatement> maker = this.visitForControl(ctx.forControl()); Statement loopBody = this.unpackStatement((Statement) this.visit(ctx.statement())); return configureAST(maker.apply(loopBody), ctx); } + /** + * Transforms {@code for await (item in source) { ... }} into a while-loop + * over an {@link groovy.concurrent.AsyncStream}: the source expression is + * adapted via {@code AsyncSupport.toAsyncStream()}, then repeatedly polled + * with {@code moveNext()} / {@code getCurrent()}. + */ + private Statement visitForAwait(final ForStmtAltContext ctx) { + ForControlContext forCtrl = ctx.forControl(); + EnhancedForControlContext enhCtrl = forCtrl.enhancedForControl(); + if (enhCtrl == null) { + throw createParsingFailedException("for await requires enhanced for syntax: for await (item in source)", ctx); + } + + ClassNode varType = this.visitType(enhCtrl.type()); + String varName = this.visitIdentifier(enhCtrl.identifier()); + Expression source = (Expression) this.visit(enhCtrl.expression()); Review Comment: `for await` uses `enhancedForControl`, which supports `(indexVariable COMMA)? variableModifiersOpt ...` (see grammar), but this desugaring ignores `enhCtrl.indexVariable()` and `enhCtrl.variableModifiersOpt()`. This means `for await (i, item in src)` and modifiers like `final`/annotations on the loop variable will compile but be dropped. Please either reject these forms with a parsing error or implement them (e.g., maintain an index counter and apply modifiers to the declared variable). ########## src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java: ########## @@ -97,6 +98,7 @@ import org.codehaus.groovy.ast.expr.RangeExpression; import org.codehaus.groovy.ast.expr.SpreadExpression; import org.codehaus.groovy.ast.expr.SpreadMapExpression; +import org.codehaus.groovy.ast.expr.StaticMethodCallExpression; Review Comment: Unused import `org.codehaus.groovy.ast.expr.StaticMethodCallExpression` will cause a Java compilation error. Remove it or use it (currently no references in this file). ```suggestion ``` ########## src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java: ########## @@ -465,14 +467,60 @@ public Statement visitLoopStmtAlt(final LoopStmtAltContext ctx) { } @Override - public ForStatement visitForStmtAlt(final ForStmtAltContext ctx) { + public Statement visitForStmtAlt(final ForStmtAltContext ctx) { + // 'for await' async iteration + if (ctx.AWAIT() != null) { + return visitForAwait(ctx); + } + Function<Statement, ForStatement> maker = this.visitForControl(ctx.forControl()); Statement loopBody = this.unpackStatement((Statement) this.visit(ctx.statement())); return configureAST(maker.apply(loopBody), ctx); } + /** + * Transforms {@code for await (item in source) { ... }} into a while-loop + * over an {@link groovy.concurrent.AsyncStream}: the source expression is + * adapted via {@code AsyncSupport.toAsyncStream()}, then repeatedly polled + * with {@code moveNext()} / {@code getCurrent()}. + */ + private Statement visitForAwait(final ForStmtAltContext ctx) { + ForControlContext forCtrl = ctx.forControl(); + EnhancedForControlContext enhCtrl = forCtrl.enhancedForControl(); + if (enhCtrl == null) { + throw createParsingFailedException("for await requires enhanced for syntax: for await (item in source)", ctx); + } + + ClassNode varType = this.visitType(enhCtrl.type()); + String varName = this.visitIdentifier(enhCtrl.identifier()); + Expression source = (Expression) this.visit(enhCtrl.expression()); + Statement loopBody = this.unpackStatement((Statement) this.visit(ctx.statement())); + + String streamVar = "$__asyncStream__" + (asyncStreamCounter++); + + // def $__asyncStream__N = AsyncSupport.toAsyncStream(source) + Expression toStreamCall = callX(AsyncTransformHelper.ASYNC_SUPPORT_TYPE, + AsyncTransformHelper.TO_ASYNC_STREAM_METHOD, new ArgumentListExpression(source)); + ExpressionStatement streamDecl = new ExpressionStatement(declX(varX(streamVar), toStreamCall)); + + // while (AsyncSupport.await($__asyncStream__N.moveNext())) + Expression moveNextCall = callX(varX(streamVar), "moveNext"); + Expression awaitCall = callX(AsyncTransformHelper.ASYNC_SUPPORT_TYPE, + AsyncTransformHelper.AWAIT_METHOD, new ArgumentListExpression(moveNextCall)); + BooleanExpression condition = new BooleanExpression(awaitCall); + + // def <varName> = $__asyncStream__N.getCurrent() + Expression getCurrentCall = callX(varX(streamVar), "getCurrent"); + ExpressionStatement getItemStmt = new ExpressionStatement(declX(varX(varName, varType), getCurrentCall)); + + BlockStatement whileBody = block(getItemStmt, loopBody); + WhileStatement whileStmt = new WhileStatement(condition, whileBody); + + return configureAST(block(streamDecl, whileStmt), ctx); Review Comment: The desugaring declares the loop variable via `declX(...)` inside the `while` body. In Groovy, the enhanced-for variable is normally declared at the loop statement level (and is visible after the loop in many contexts). This translation changes scoping/visibility and may break code that expects the loop variable to exist after the `for await`. Consider declaring the variable once before the loop and using an assignment inside the loop body (matching `visitEnhancedForControl`/`ForStatement` behavior). ```suggestion // def <varName> // declare loop variable once, before the while-loop ExpressionStatement loopVarDecl = new ExpressionStatement( declX(varX(varName, varType), ConstantExpression.NULL)); // while (AsyncSupport.await($__asyncStream__N.moveNext())) Expression moveNextCall = callX(varX(streamVar), "moveNext"); Expression awaitCall = callX(AsyncTransformHelper.ASYNC_SUPPORT_TYPE, AsyncTransformHelper.AWAIT_METHOD, new ArgumentListExpression(moveNextCall)); BooleanExpression condition = new BooleanExpression(awaitCall); // <varName> = $__asyncStream__N.getCurrent() Expression getCurrentCall = callX(varX(streamVar), "getCurrent"); ExpressionStatement getItemStmt = new ExpressionStatement( assignX(varX(varName, varType), getCurrentCall)); BlockStatement whileBody = block(getItemStmt, loopBody); WhileStatement whileStmt = new WhileStatement(condition, whileBody); return configureAST(block(streamDecl, loopVarDecl, whileStmt), ctx); ``` ########## src/test/groovy/org/codehaus/groovy/transform/AsyncVirtualThreadTest.groovy: ########## @@ -0,0 +1,1138 @@ +/* + * 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.junit.jupiter.api.Test + +import static groovy.test.GroovyAssert.assertScript + +/** + * Tests for virtual thread integration, executor configuration, exception + * handling consistency across async methods/closures/lambdas, and coverage + * improvements for the async/await feature. + * + * @since 6.0.0 + */ +final class AsyncVirtualThreadTest { + + // ---- Virtual thread detection and usage ---- + + @Test + void testVirtualThreadsAvailable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + // JDK 21+ should have virtual threads + def jdkVersion = Runtime.version().feature() + if (jdkVersion >= 21) { + assert isVirtualThreadsAvailable() + } + ''' + } + + @Test + void testAsyncMethodRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class VTService { + async checkThread() { + return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def service = new VTService() + def result = service.checkThread().get() + assert result == true + } + ''' + } + + @Test + void testAsyncClosureRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncVirtual = async { Thread.currentThread().isVirtual() } + def awaitable = asyncVirtual() + assert await(awaitable) == true + } + ''' + } + + @Test + void testAsyncLambdaRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncFn = async { x -> Thread.currentThread().isVirtual() } + def result = await(asyncFn(42)) + assert result == true + } + ''' + } + + @Test + void testForAwaitRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.AsyncStream + + class VTGenerator { + async generate() { + yield return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def gen = new VTGenerator() + def stream = gen.generate() + def results = [] + for await (item in stream) { + results << item + } + assert results == [true] + } + ''' + } + + @Test + void testHighConcurrencyWithVirtualThreads() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class HighConcurrency { + async compute(int n) { + Thread.sleep(10) + return n * 2 + } + } + + if (isVirtualThreadsAvailable()) { + def svc = new HighConcurrency() + // Launch 1000 concurrent tasks — trivial with virtual threads + def awaitables = (1..1000).collect { svc.compute(it) } + def results = awaitAll(awaitables as Object[]) + assert results.size() == 1000 + assert results[0] == 2 + assert results[999] == 2000 + } + ''' + } + + // ---- Executor configuration ---- + + @Test + void testCustomExecutorOverride() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.atomic.AtomicReference + + def savedExecutor = getExecutor() + try { + def customPool = Executors.newFixedThreadPool(2, { r -> + def t = new Thread(r) + t.setName("custom-async-" + t.getId()) + t + }) + setExecutor(customPool) + + def asyncName = async { + Thread.currentThread().getName() + } + def awaitable = asyncName() + def threadName = await(awaitable) + assert threadName.startsWith("custom-async-") + } finally { + setExecutor(savedExecutor) + } + ''' + } + + @Test + void testSetExecutorNullResetsToDefault() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + + def originalExecutor = getExecutor() + // Set a custom executor + setExecutor(Executors.newSingleThreadExecutor()) + assert getExecutor() != originalExecutor + // Reset to null — should restore default + setExecutor(null) + def restored = getExecutor() + assert restored != null + // Verify it works + def awaitable = groovy.concurrent.AsyncUtils.async { 42 } + assert groovy.concurrent.AsyncUtils.await(awaitable) == 42 + // Restore original + setExecutor(originalExecutor) + ''' + } + + @Test + void testAsyncMethodWithCustomExecutorField() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executor + import java.util.concurrent.Executors + + class CustomExecutorService { + static Executor myPool = Executors.newFixedThreadPool(1, { r -> + def t = new Thread(r) + t.setName("my-pool-thread") + t + }) + + @Async(executor = "myPool") + def doWork() { + return Thread.currentThread().getName() + } + } + + def svc = new CustomExecutorService() + def result = svc.doWork().get() + assert result.startsWith("my-pool-thread") + ''' + } + + // ---- Exception handling consistency: async method vs closure vs lambda ---- + + @Test + void testCheckedExceptionConsistencyAcrossAsyncMethods() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class ExcService { + async failWithChecked() { + throw new java.io.IOException("async method error") + } + } + + def svc = new ExcService() + try { + await(svc.failWithChecked()) + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message == "async method error" + } + ''' + } + + @Test + void testCheckedExceptionConsistencyAcrossClosures() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncIOError = async { throw new java.io.IOException("closure error") } + def awaitable = asyncIOError() + try { + await(awaitable) + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message == "closure error" + } + ''' + } + + @Test + void testCheckedExceptionConsistencyAcrossLambdas() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncFn = async { x -> throw new java.io.IOException("lambda error: ${x}") } + try { + await(asyncFn("test")) + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message == "lambda error: test" + } + ''' + } + + @Test + void testRuntimeExceptionConsistencyAllForms() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + // Form 1: async method + class Svc { + async failMethod() { throw new IllegalStateException("from method") } + } + + // Form 2: async closure + def asyncClosure = async { throw new IllegalArgumentException("from closure") } + def closure = asyncClosure() + + // Form 3: async lambda with params + def lambda = async { x -> throw new UnsupportedOperationException("from lambda") } + + // All should throw the exact exception type (no wrapping) + try { await(new Svc().failMethod()); assert false } + catch (IllegalStateException e) { assert e.message == "from method" } + + try { await(closure); assert false } + catch (IllegalArgumentException e) { assert e.message == "from closure" } + + try { await(lambda("x")); assert false } + catch (UnsupportedOperationException e) { assert e.message == "from lambda" } + ''' + } + + @Test + void testErrorPropagationConsistencyAllForms() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class ErrorSvc { + async fail() { throw new StackOverflowError("method") } + } + + // async method + try { await(new ErrorSvc().fail()); assert false } + catch (StackOverflowError e) { assert e.message == "method" } + + // async closure + def asyncSOE = async { throw new StackOverflowError("closure") } + try { await(asyncSOE()); assert false } + catch (StackOverflowError e) { assert e.message == "closure" } + + // async lambda + def fn = async { x -> throw new StackOverflowError("lambda") } + try { await(fn(1)); assert false } + catch (StackOverflowError e) { assert e.message == "lambda" } + ''' + } + + // ---- Async stream (yield return) consistency ---- + + @Test + void testYieldReturnConsistencyMethodVsClosure() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + // Form 1: async method with yield return + class GenSvc { + async range(int n) { + for (int i = 1; i <= n; i++) { + yield return i + } + } + } + + // Form 2: async closure with yield return + def asyncClosureGen = async { for (int i = 1; i <= 3; i++) { yield return i * 10 } } + def closureGen = asyncClosureGen() + + // Form 3: async lambda with yield return + params + def lambdaGen = async { n -> for (int i = 1; i <= n; i++) { yield return i * 100 } } + + // Verify all produce correct streams + def methodResults = [] + for await (item in new GenSvc().range(3)) { methodResults << item } + assert methodResults == [1, 2, 3] + + def closureResults = [] + for await (item in closureGen) { closureResults << item } + assert closureResults == [10, 20, 30] + + def lambdaResults = [] + for await (item in lambdaGen(3)) { lambdaResults << item } + assert lambdaResults == [100, 200, 300] + ''' + } + + @Test + void testYieldReturnExceptionConsistencyAllForms() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + // async method generator with error + class FailGen { + async failing() { + yield return 1 + throw new java.io.IOException("gen method error") + } + } + + // async closure generator with error + def asyncFailGen = async { + yield return 10 + throw new java.io.IOException("gen closure error") + } + def closureFailGen = asyncFailGen() + + // async lambda generator with error + def lambdaFailGen = async { x -> + yield return x + throw new java.io.IOException("gen lambda error") + } + + // All should propagate IOException through for-await + for (gen in [new FailGen().failing(), closureFailGen, lambdaFailGen(100)]) { + def items = [] + try { + for await (item in gen) { items << item } + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message.contains("gen") + assert items.size() == 1 + } + } + ''' + } + + // ---- executeAsync / executeAsyncVoid (unified path) ---- + + @Test + void testExecuteAsyncWithCustomExecutor() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + + def pool = Executors.newSingleThreadExecutor({ r -> + def t = new Thread(r) + t.setName("exec-async-test") + t + }) + + def awaitable = executeAsync({ -> + Thread.currentThread().getName() + }, pool) + def result = await(awaitable) + assert result.startsWith("exec-async-test") + pool.shutdown() + ''' + } + + @Test + void testExecuteAsyncVoidWithCustomExecutor() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.atomic.AtomicBoolean + + def pool = Executors.newSingleThreadExecutor() + def executed = new AtomicBoolean(false) + + def awaitable = executeAsyncVoid({ -> + executed.set(true) + }, pool) + def _v1 = await(awaitable) + assert executed.get() + pool.shutdown() + ''' + } + + @Test + void testAsyncVoidMethodReturnsAwaitable() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + import java.util.concurrent.atomic.AtomicInteger + + class VoidService { + static AtomicInteger counter = new AtomicInteger(0) + + @Async + void increment() { + counter.incrementAndGet() + } + } + + def svc = new VoidService() + def awaitable = svc.increment() + assert awaitable instanceof Awaitable + def _v2 = await(awaitable) + assert VoidService.counter.get() == 1 + ''' + } + + // ---- Edge cases and coverage improvements ---- + + @Test + void testAwaitNull() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + def result = await((Object) null) + assert result == null + ''' + } + + @Test + void testAwaitAlreadyCompletedAwaitable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def awaitable = Awaitable.of(42) + assert awaitable.isDone() + def result = await(awaitable) + assert result == 42 + ''' + } + + @Test + void testAwaitFailedAwaitable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def awaitable = Awaitable.failed(new RuntimeException("pre-failed")) + assert awaitable.isCompletedExceptionally() + try { + await(awaitable) + assert false + } catch (RuntimeException e) { + assert e.message == "pre-failed" + } + ''' + } + + @Test + void testAwaitCancelledAwaitable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import org.apache.groovy.runtime.async.GroovyPromise + import java.util.concurrent.CompletableFuture + import java.util.concurrent.CancellationException + + def cf = new CompletableFuture() + def awaitable = GroovyPromise.of(cf) + awaitable.cancel() + assert awaitable.isCancelled() + try { + await(awaitable) + assert false + } catch (CancellationException e) { + // expected + } + ''' + } + + @Test + void testAsyncWithReturnValue() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncHello = async { return "hello" } + def awaitable = asyncHello() + assert await(awaitable) == "hello" + ''' + } + + @Test + void testAsyncWithNullReturnValue() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncNull = async { return null } + def awaitable = asyncNull() + assert await(awaitable) == null + ''' + } + + @Test + void testMultipleConcurrentAsyncGenerators() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncGen1 = async { for (int i = 0; i < 5; i++) { yield return "A${i}" } } + def gen1 = asyncGen1() + def asyncGen2 = async { for (int i = 0; i < 5; i++) { yield return "B${i}" } } + def gen2 = asyncGen2() + + def results1 = [] + def results2 = [] + for await (item in gen1) { results1 << item } + for await (item in gen2) { results2 << item } + assert results1 == ["A0", "A1", "A2", "A3", "A4"] + assert results2 == ["B0", "B1", "B2", "B3", "B4"] + ''' + } + + @Test + void testEmptyAsyncStream() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class EmptyGen { + async empty() { + // A generator must have yield return to be detected as one. + // This has one but it's unreachable at runtime. + if (false) yield return "unreachable" + } + } + + def results = [] + for await (item in new EmptyGen().empty()) { + results << item + } + assert results.isEmpty() + ''' + } + + @Test + void testAwaitableCompositionWithThen() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def asyncOriginal = async { 10 } + def original = asyncOriginal() + def doubled = original.then { it * 2 } + def result = await(doubled) + assert result == 20 + ''' + } + + @Test + void testAwaitableCompositionWithThenCompose() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def asyncFirst = async { 5 } + def first = asyncFirst() + def chained = first.thenCompose { v -> def asyncCompose = async { v + 10 }; asyncCompose() } + def result = await(chained) + assert result == 15 + ''' + } + + @Test + void testAwaitableExceptionally() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def asyncFailing = async { throw new RuntimeException("oops") } + def failing = asyncFailing() + def recovered = failing.exceptionally { e -> "recovered: ${e.message}" } + def result = await(recovered) + assert result == "recovered: oops" + ''' + } + + @Test + void testAwaitWithCompletionStage() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.CompletableFuture + import java.util.concurrent.CompletionStage + + CompletionStage stage = CompletableFuture.supplyAsync { "from stage" } + def result = await(stage) + assert result == "from stage" + ''' + } + + @Test + void testAwaitWithFuture() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.Future + + def pool = Executors.newSingleThreadExecutor() + Future future = pool.submit({ "from future" } as java.util.concurrent.Callable) + def result = await(future) + assert result == "from future" + pool.shutdown() + ''' + } + + @Test + void testAwaitAllWithMixedTypes() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + import java.util.concurrent.CompletableFuture + + def asyncA1 = async { 1 } + def a1 = asyncA1() + def a2 = CompletableFuture.supplyAsync { 2 } + def a3 = Awaitable.of(3) + + def results = awaitAll(a1, a2, a3) + assert results == [1, 2, 3] + ''' + } + + @Test + void testAwaitAnyReturnsFirst() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def fast = Awaitable.of("fast") + def asyncSlow = async { Thread.sleep(500); "slow" } + def slow = asyncSlow() + + def result = awaitAny(fast, slow) + assert result == "fast" + ''' + } + + @Test + void testAwaitAllSettledMixed() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncSuccess = async { "ok" } + def success = asyncSuccess() + def asyncFailure = async { throw new RuntimeException("fail") } + def failure = asyncFailure() + + def results = awaitAllSettled(success, failure) + assert results.size() == 2 + assert results[0].isSuccess() + assert results[0].getValue() == "ok" + assert results[1].isFailure() + assert results[1].getError().message == "fail" + ''' + } + + @Test + void testAsyncMethodWithPrimitiveReturnType() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class PrimService { + @Async + int computeInt() { return 42 } + + @Async + boolean checkBool() { return true } + + @Async + double computeDouble() { return 3.14 } + } + + def svc = new PrimService() + assert await(svc.computeInt()) == 42 + assert await(svc.checkBool()) == true + assert Math.abs(await(svc.computeDouble()) - 3.14) < 0.001 + ''' + } + + @Test + void testDeepUnwrapNestedExceptions() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.CompletionException + import java.util.concurrent.ExecutionException + + // Create deeply nested exception chain + def original = new java.io.IOException("deep") + def wrapped = new CompletionException(new ExecutionException( + new java.lang.reflect.UndeclaredThrowableException( + new java.lang.reflect.InvocationTargetException(original)))) + + def cf = new java.util.concurrent.CompletableFuture() + cf.completeExceptionally(wrapped) + + try { + await(cf) + assert false + } catch (java.io.IOException e) { + assert e.message == "deep" + } + ''' + } + + @Test + void testAsyncWithNestedAwait() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class NestedService { + async inner(int v) { return v * 2 } + + async outer(int v) { + def intermediate = await inner(v) + return await inner(intermediate) + } + } + + def svc = new NestedService() + def result = await(svc.outer(5)) + assert result == 20 // 5 * 2 * 2 + ''' + } + + @Test + void testAsyncClosureWithNestedAwait() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def inner = async { x -> x + 1 } + def asyncOuter = async { + def v1 = await(inner(10)) + def v2 = await(inner(v1)) + return v2 + } + def outer = asyncOuter() + assert await(outer) == 12 // 10 + 1 + 1 + ''' + } + + @Test + void testParallelAwaitInAsyncMethod() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.CompletableFuture + + class ParallelService { + CompletableFuture<Integer> compute(int v) { + return CompletableFuture.supplyAsync { + Thread.sleep(50) + return v * 2 + } + } + + async parallel() { + def f1 = compute(1) + def f2 = compute(2) + def f3 = compute(3) + return await(f1) + await(f2) + await(f3) + } + } + + def svc = new ParallelService() + long start = System.currentTimeMillis() + def result = await(svc.parallel()) + long elapsed = System.currentTimeMillis() - start + assert result == 12 // 2 + 4 + 6 + assert elapsed < 500 // parallel, not sequential Review Comment: This test asserts a wall-clock threshold (`elapsed < 500`) to prove parallelism. Timing-based assertions are prone to CI flakiness under load/GC pauses and can fail even when behavior is correct. Consider using a deterministic signal (latches/barriers) or relaxing/removing the hard time bound. ```suggestion def result = await(svc.parallel()) assert result == 12 // 2 + 4 + 6 ``` ########## src/antlr/GroovyParser.g4: ########## @@ -134,6 +134,7 @@ modifier | VOLATILE | DEF | VAR + | ASYNC ) Review Comment: Adding `ASYNC` to the generic `modifier` rule makes `async` a legal modifier on many declarations (fields, classes, etc.), but the implementation only gives it meaning for method declarations (via annotation injection in AstBuilder). As-is, `async` on non-method constructs will likely be silently accepted but have no effect; consider restricting `ASYNC` to method modifiers in the grammar, or emitting a clear parsing/semantic error when used on unsupported targets. ```suggestion ) | { getContext() instanceof GroovyParser.MethodDeclarationContext }? m=ASYNC ``` ########## src/test/groovy/org/codehaus/groovy/transform/AsyncVirtualThreadTest.groovy: ########## @@ -0,0 +1,1138 @@ +/* + * 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.junit.jupiter.api.Test + +import static groovy.test.GroovyAssert.assertScript + +/** + * Tests for virtual thread integration, executor configuration, exception + * handling consistency across async methods/closures/lambdas, and coverage + * improvements for the async/await feature. + * + * @since 6.0.0 + */ +final class AsyncVirtualThreadTest { + + // ---- Virtual thread detection and usage ---- + + @Test + void testVirtualThreadsAvailable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + // JDK 21+ should have virtual threads + def jdkVersion = Runtime.version().feature() + if (jdkVersion >= 21) { + assert isVirtualThreadsAvailable() + } + ''' + } + + @Test + void testAsyncMethodRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class VTService { + async checkThread() { + return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def service = new VTService() + def result = service.checkThread().get() + assert result == true + } + ''' + } + + @Test + void testAsyncClosureRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncVirtual = async { Thread.currentThread().isVirtual() } + def awaitable = asyncVirtual() + assert await(awaitable) == true + } + ''' + } + + @Test + void testAsyncLambdaRunsOnVirtualThread() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + if (isVirtualThreadsAvailable()) { + def asyncFn = async { x -> Thread.currentThread().isVirtual() } + def result = await(asyncFn(42)) + assert result == true + } + ''' + } + + @Test + void testForAwaitRunsOnVirtualThread() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.AsyncStream + + class VTGenerator { + async generate() { + yield return Thread.currentThread().isVirtual() + } + } + + if (isVirtualThreadsAvailable()) { + def gen = new VTGenerator() + def stream = gen.generate() + def results = [] + for await (item in stream) { + results << item + } + assert results == [true] + } + ''' + } + + @Test + void testHighConcurrencyWithVirtualThreads() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class HighConcurrency { + async compute(int n) { + Thread.sleep(10) + return n * 2 + } + } + + if (isVirtualThreadsAvailable()) { + def svc = new HighConcurrency() + // Launch 1000 concurrent tasks — trivial with virtual threads + def awaitables = (1..1000).collect { svc.compute(it) } + def results = awaitAll(awaitables as Object[]) + assert results.size() == 1000 + assert results[0] == 2 + assert results[999] == 2000 + } + ''' + } + + // ---- Executor configuration ---- + + @Test + void testCustomExecutorOverride() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.atomic.AtomicReference + + def savedExecutor = getExecutor() + try { + def customPool = Executors.newFixedThreadPool(2, { r -> + def t = new Thread(r) + t.setName("custom-async-" + t.getId()) + t + }) + setExecutor(customPool) + + def asyncName = async { + Thread.currentThread().getName() + } + def awaitable = asyncName() + def threadName = await(awaitable) + assert threadName.startsWith("custom-async-") + } finally { + setExecutor(savedExecutor) + } + ''' + } + + @Test + void testSetExecutorNullResetsToDefault() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + + def originalExecutor = getExecutor() + // Set a custom executor + setExecutor(Executors.newSingleThreadExecutor()) + assert getExecutor() != originalExecutor + // Reset to null — should restore default + setExecutor(null) + def restored = getExecutor() + assert restored != null + // Verify it works + def awaitable = groovy.concurrent.AsyncUtils.async { 42 } + assert groovy.concurrent.AsyncUtils.await(awaitable) == 42 + // Restore original + setExecutor(originalExecutor) + ''' + } + + @Test + void testAsyncMethodWithCustomExecutorField() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executor + import java.util.concurrent.Executors + + class CustomExecutorService { + static Executor myPool = Executors.newFixedThreadPool(1, { r -> + def t = new Thread(r) + t.setName("my-pool-thread") + t + }) + + @Async(executor = "myPool") + def doWork() { + return Thread.currentThread().getName() + } + } + + def svc = new CustomExecutorService() + def result = svc.doWork().get() + assert result.startsWith("my-pool-thread") + ''' + } + + // ---- Exception handling consistency: async method vs closure vs lambda ---- + + @Test + void testCheckedExceptionConsistencyAcrossAsyncMethods() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class ExcService { + async failWithChecked() { + throw new java.io.IOException("async method error") + } + } + + def svc = new ExcService() + try { + await(svc.failWithChecked()) + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message == "async method error" + } + ''' + } + + @Test + void testCheckedExceptionConsistencyAcrossClosures() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncIOError = async { throw new java.io.IOException("closure error") } + def awaitable = asyncIOError() + try { + await(awaitable) + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message == "closure error" + } + ''' + } + + @Test + void testCheckedExceptionConsistencyAcrossLambdas() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncFn = async { x -> throw new java.io.IOException("lambda error: ${x}") } + try { + await(asyncFn("test")) + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message == "lambda error: test" + } + ''' + } + + @Test + void testRuntimeExceptionConsistencyAllForms() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + // Form 1: async method + class Svc { + async failMethod() { throw new IllegalStateException("from method") } + } + + // Form 2: async closure + def asyncClosure = async { throw new IllegalArgumentException("from closure") } + def closure = asyncClosure() + + // Form 3: async lambda with params + def lambda = async { x -> throw new UnsupportedOperationException("from lambda") } + + // All should throw the exact exception type (no wrapping) + try { await(new Svc().failMethod()); assert false } + catch (IllegalStateException e) { assert e.message == "from method" } + + try { await(closure); assert false } + catch (IllegalArgumentException e) { assert e.message == "from closure" } + + try { await(lambda("x")); assert false } + catch (UnsupportedOperationException e) { assert e.message == "from lambda" } + ''' + } + + @Test + void testErrorPropagationConsistencyAllForms() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class ErrorSvc { + async fail() { throw new StackOverflowError("method") } + } + + // async method + try { await(new ErrorSvc().fail()); assert false } + catch (StackOverflowError e) { assert e.message == "method" } + + // async closure + def asyncSOE = async { throw new StackOverflowError("closure") } + try { await(asyncSOE()); assert false } + catch (StackOverflowError e) { assert e.message == "closure" } + + // async lambda + def fn = async { x -> throw new StackOverflowError("lambda") } + try { await(fn(1)); assert false } + catch (StackOverflowError e) { assert e.message == "lambda" } + ''' + } + + // ---- Async stream (yield return) consistency ---- + + @Test + void testYieldReturnConsistencyMethodVsClosure() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + // Form 1: async method with yield return + class GenSvc { + async range(int n) { + for (int i = 1; i <= n; i++) { + yield return i + } + } + } + + // Form 2: async closure with yield return + def asyncClosureGen = async { for (int i = 1; i <= 3; i++) { yield return i * 10 } } + def closureGen = asyncClosureGen() + + // Form 3: async lambda with yield return + params + def lambdaGen = async { n -> for (int i = 1; i <= n; i++) { yield return i * 100 } } + + // Verify all produce correct streams + def methodResults = [] + for await (item in new GenSvc().range(3)) { methodResults << item } + assert methodResults == [1, 2, 3] + + def closureResults = [] + for await (item in closureGen) { closureResults << item } + assert closureResults == [10, 20, 30] + + def lambdaResults = [] + for await (item in lambdaGen(3)) { lambdaResults << item } + assert lambdaResults == [100, 200, 300] + ''' + } + + @Test + void testYieldReturnExceptionConsistencyAllForms() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + // async method generator with error + class FailGen { + async failing() { + yield return 1 + throw new java.io.IOException("gen method error") + } + } + + // async closure generator with error + def asyncFailGen = async { + yield return 10 + throw new java.io.IOException("gen closure error") + } + def closureFailGen = asyncFailGen() + + // async lambda generator with error + def lambdaFailGen = async { x -> + yield return x + throw new java.io.IOException("gen lambda error") + } + + // All should propagate IOException through for-await + for (gen in [new FailGen().failing(), closureFailGen, lambdaFailGen(100)]) { + def items = [] + try { + for await (item in gen) { items << item } + assert false : "Should have thrown" + } catch (java.io.IOException e) { + assert e.message.contains("gen") + assert items.size() == 1 + } + } + ''' + } + + // ---- executeAsync / executeAsyncVoid (unified path) ---- + + @Test + void testExecuteAsyncWithCustomExecutor() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + + def pool = Executors.newSingleThreadExecutor({ r -> + def t = new Thread(r) + t.setName("exec-async-test") + t + }) + + def awaitable = executeAsync({ -> + Thread.currentThread().getName() + }, pool) + def result = await(awaitable) + assert result.startsWith("exec-async-test") + pool.shutdown() + ''' + } + + @Test + void testExecuteAsyncVoidWithCustomExecutor() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.atomic.AtomicBoolean + + def pool = Executors.newSingleThreadExecutor() + def executed = new AtomicBoolean(false) + + def awaitable = executeAsyncVoid({ -> + executed.set(true) + }, pool) + def _v1 = await(awaitable) + assert executed.get() + pool.shutdown() + ''' + } + + @Test + void testAsyncVoidMethodReturnsAwaitable() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + import java.util.concurrent.atomic.AtomicInteger + + class VoidService { + static AtomicInteger counter = new AtomicInteger(0) + + @Async + void increment() { + counter.incrementAndGet() + } + } + + def svc = new VoidService() + def awaitable = svc.increment() + assert awaitable instanceof Awaitable + def _v2 = await(awaitable) + assert VoidService.counter.get() == 1 + ''' + } + + // ---- Edge cases and coverage improvements ---- + + @Test + void testAwaitNull() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + def result = await((Object) null) + assert result == null + ''' + } + + @Test + void testAwaitAlreadyCompletedAwaitable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def awaitable = Awaitable.of(42) + assert awaitable.isDone() + def result = await(awaitable) + assert result == 42 + ''' + } + + @Test + void testAwaitFailedAwaitable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def awaitable = Awaitable.failed(new RuntimeException("pre-failed")) + assert awaitable.isCompletedExceptionally() + try { + await(awaitable) + assert false + } catch (RuntimeException e) { + assert e.message == "pre-failed" + } + ''' + } + + @Test + void testAwaitCancelledAwaitable() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import org.apache.groovy.runtime.async.GroovyPromise + import java.util.concurrent.CompletableFuture + import java.util.concurrent.CancellationException + + def cf = new CompletableFuture() + def awaitable = GroovyPromise.of(cf) + awaitable.cancel() + assert awaitable.isCancelled() + try { + await(awaitable) + assert false + } catch (CancellationException e) { + // expected + } + ''' + } + + @Test + void testAsyncWithReturnValue() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncHello = async { return "hello" } + def awaitable = asyncHello() + assert await(awaitable) == "hello" + ''' + } + + @Test + void testAsyncWithNullReturnValue() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncNull = async { return null } + def awaitable = asyncNull() + assert await(awaitable) == null + ''' + } + + @Test + void testMultipleConcurrentAsyncGenerators() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncGen1 = async { for (int i = 0; i < 5; i++) { yield return "A${i}" } } + def gen1 = asyncGen1() + def asyncGen2 = async { for (int i = 0; i < 5; i++) { yield return "B${i}" } } + def gen2 = asyncGen2() + + def results1 = [] + def results2 = [] + for await (item in gen1) { results1 << item } + for await (item in gen2) { results2 << item } + assert results1 == ["A0", "A1", "A2", "A3", "A4"] + assert results2 == ["B0", "B1", "B2", "B3", "B4"] + ''' + } + + @Test + void testEmptyAsyncStream() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class EmptyGen { + async empty() { + // A generator must have yield return to be detected as one. + // This has one but it's unreachable at runtime. + if (false) yield return "unreachable" + } + } + + def results = [] + for await (item in new EmptyGen().empty()) { + results << item + } + assert results.isEmpty() + ''' + } + + @Test + void testAwaitableCompositionWithThen() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def asyncOriginal = async { 10 } + def original = asyncOriginal() + def doubled = original.then { it * 2 } + def result = await(doubled) + assert result == 20 + ''' + } + + @Test + void testAwaitableCompositionWithThenCompose() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def asyncFirst = async { 5 } + def first = asyncFirst() + def chained = first.thenCompose { v -> def asyncCompose = async { v + 10 }; asyncCompose() } + def result = await(chained) + assert result == 15 + ''' + } + + @Test + void testAwaitableExceptionally() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def asyncFailing = async { throw new RuntimeException("oops") } + def failing = asyncFailing() + def recovered = failing.exceptionally { e -> "recovered: ${e.message}" } + def result = await(recovered) + assert result == "recovered: oops" + ''' + } + + @Test + void testAwaitWithCompletionStage() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.CompletableFuture + import java.util.concurrent.CompletionStage + + CompletionStage stage = CompletableFuture.supplyAsync { "from stage" } + def result = await(stage) + assert result == "from stage" + ''' + } + + @Test + void testAwaitWithFuture() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.Executors + import java.util.concurrent.Future + + def pool = Executors.newSingleThreadExecutor() + Future future = pool.submit({ "from future" } as java.util.concurrent.Callable) + def result = await(future) + assert result == "from future" + pool.shutdown() + ''' + } + + @Test + void testAwaitAllWithMixedTypes() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + import java.util.concurrent.CompletableFuture + + def asyncA1 = async { 1 } + def a1 = asyncA1() + def a2 = CompletableFuture.supplyAsync { 2 } + def a3 = Awaitable.of(3) + + def results = awaitAll(a1, a2, a3) + assert results == [1, 2, 3] + ''' + } + + @Test + void testAwaitAnyReturnsFirst() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + def fast = Awaitable.of("fast") + def asyncSlow = async { Thread.sleep(500); "slow" } + def slow = asyncSlow() + + def result = awaitAny(fast, slow) + assert result == "fast" + ''' + } + + @Test + void testAwaitAllSettledMixed() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def asyncSuccess = async { "ok" } + def success = asyncSuccess() + def asyncFailure = async { throw new RuntimeException("fail") } + def failure = asyncFailure() + + def results = awaitAllSettled(success, failure) + assert results.size() == 2 + assert results[0].isSuccess() + assert results[0].getValue() == "ok" + assert results[1].isFailure() + assert results[1].getError().message == "fail" + ''' + } + + @Test + void testAsyncMethodWithPrimitiveReturnType() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class PrimService { + @Async + int computeInt() { return 42 } + + @Async + boolean checkBool() { return true } + + @Async + double computeDouble() { return 3.14 } + } + + def svc = new PrimService() + assert await(svc.computeInt()) == 42 + assert await(svc.checkBool()) == true + assert Math.abs(await(svc.computeDouble()) - 3.14) < 0.001 + ''' + } + + @Test + void testDeepUnwrapNestedExceptions() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.CompletionException + import java.util.concurrent.ExecutionException + + // Create deeply nested exception chain + def original = new java.io.IOException("deep") + def wrapped = new CompletionException(new ExecutionException( + new java.lang.reflect.UndeclaredThrowableException( + new java.lang.reflect.InvocationTargetException(original)))) + + def cf = new java.util.concurrent.CompletableFuture() + cf.completeExceptionally(wrapped) + + try { + await(cf) + assert false + } catch (java.io.IOException e) { + assert e.message == "deep" + } + ''' + } + + @Test + void testAsyncWithNestedAwait() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class NestedService { + async inner(int v) { return v * 2 } + + async outer(int v) { + def intermediate = await inner(v) + return await inner(intermediate) + } + } + + def svc = new NestedService() + def result = await(svc.outer(5)) + assert result == 20 // 5 * 2 * 2 + ''' + } + + @Test + void testAsyncClosureWithNestedAwait() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def inner = async { x -> x + 1 } + def asyncOuter = async { + def v1 = await(inner(10)) + def v2 = await(inner(v1)) + return v2 + } + def outer = asyncOuter() + assert await(outer) == 12 // 10 + 1 + 1 + ''' + } + + @Test + void testParallelAwaitInAsyncMethod() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.CompletableFuture + + class ParallelService { + CompletableFuture<Integer> compute(int v) { + return CompletableFuture.supplyAsync { + Thread.sleep(50) + return v * 2 + } + } + + async parallel() { + def f1 = compute(1) + def f2 = compute(2) + def f3 = compute(3) + return await(f1) + await(f2) + await(f3) + } + } + + def svc = new ParallelService() + long start = System.currentTimeMillis() + def result = await(svc.parallel()) + long elapsed = System.currentTimeMillis() - start + assert result == 12 // 2 + 4 + 6 + assert elapsed < 500 // parallel, not sequential + ''' + } + + @Test + void testToAsyncStreamConversion() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.AsyncStream + import groovy.concurrent.Awaitable + import groovy.concurrent.AwaitableAdapter + import groovy.concurrent.AwaitableAdapterRegistry + + // Register an adapter for List that provides an AsyncStream + AwaitableAdapterRegistry.register(new AwaitableAdapter() { + boolean supportsAwaitable(Class type) { return false } + Awaitable toAwaitable(Object source) { return null } + boolean supportsAsyncStream(Class type) { return List.isAssignableFrom(type) } + AsyncStream toAsyncStream(Object source) { + def list = (List) source + def iter = list.iterator() + return new AsyncStream() { + def current + Awaitable<Boolean> moveNext() { + if (iter.hasNext()) { + current = iter.next() + return Awaitable.of(true) + } + return Awaitable.of(false) + } + Object getCurrent() { return current } + } + } + }) + + def results = [] + def stream = toAsyncStream([10, 20, 30]) + for await (item in stream) { + results << item + } + assert results == [10, 20, 30] + ''' + } + + @Test + void testAwaitObjectDispatchesToCorrectOverload() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + import java.util.concurrent.CompletableFuture + + // Test Object overload dispatch for each type + Object a = Awaitable.of("awaitable") + Object b = CompletableFuture.completedFuture("cf") + Object c = CompletableFuture.completedFuture("stage") as java.util.concurrent.CompletionStage + + assert await(a) == "awaitable" + assert await(b) == "cf" + assert await(c) == "stage" + ''' + } + + @Test + void testAsyncMethodWithParameters() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class ParamService { + async add(int a, int b) { + return a + b + } + + async greet(String name) { + return "Hello, ${name}!" + } + } + + def svc = new ParamService() + assert await(svc.add(3, 4)) == 7 + assert await(svc.greet("Groovy")) == "Hello, Groovy!" + ''' + } + + @Test + void testAsyncChainedMethods() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + class Pipeline { + async step1() { return 1 } + + async step2(Awaitable<Integer> input) { + return await(input) + 10 + } + + async step3(Awaitable<Integer> input) { + return await(input) * 2 + } + } + + def p = new Pipeline() + def r1 = p.step1() + def r2 = p.step2(r1) + def r3 = p.step3(r2) + assert r3.get() == 22 + ''' + } + + @Test + void testAwaitSyntaxWithBinaryOperations() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + import groovy.concurrent.Awaitable + + class MathService { + async compute(int v) { return v } + } + + def svc = new MathService() + // Test that await properly integrates with binary operators + def a = svc.compute(10) + def b = svc.compute(20) + def r1 = await(a) + def r2 = await(b) + assert r1 + r2 == 30 + ''' + } + + @Test + void testAsyncClosureWithoutYieldReturnIsNotGenerator() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + // Closure with no yield return produces an Awaitable (not AsyncStream) + def asyncResult = async { 42 } + def result = asyncResult() + assert await(result) == 42 + + def asyncResultNull = async { /* empty body */ } + def resultNull = asyncResultNull() + assert await(resultNull) == null + ''' + } + + @Test + void testGroovyPromiseToString() { + assertScript ''' + import org.apache.groovy.runtime.async.GroovyPromise + import groovy.concurrent.Awaitable + import java.util.concurrent.CompletableFuture + + def pending = new CompletableFuture() + def promise = GroovyPromise.of(pending) + assert promise.toString() == "GroovyPromise{pending}" + + pending.complete(42) + assert promise.toString() == "GroovyPromise{completed}" + + def failed = GroovyPromise.of(CompletableFuture.failedFuture(new RuntimeException())) + assert failed.toString() == "GroovyPromise{failed}" + ''' + } + + @Test + void testGroovyPromiseGetWithTimeout() { + assertScript ''' + import org.apache.groovy.runtime.async.GroovyPromise + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.TimeUnit + import java.util.concurrent.TimeoutException + + def asyncTimeout = async { Thread.sleep(5000); "never" } Review Comment: This test starts an async task that sleeps for 5 seconds but only asserts a 50ms timeout on `get()`. The underlying task will continue sleeping after the timeout, which can unnecessarily consume executor threads and slow/flakify the overall suite (especially on the fixed-thread fallback). Consider using a much shorter sleep, cancelling the awaitable after the timeout, or using `delay(...)` with a cancellation-aware mechanism. ```suggestion def asyncTimeout = async { Thread.sleep(100); "never" } ``` ########## src/test/groovy/org/codehaus/groovy/transform/AsyncPatternsTest.groovy: ########## @@ -0,0 +1,1006 @@ +/* + * 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 groovy.concurrent.AsyncUtils +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Test + +import static groovy.test.GroovyAssert.assertScript + +/** + * Tests covering async/await patterns including CPU-bound work, I/O-bound + * simulation, caller perspectives, continuations, parallel execution, + * delay, exception handling, async closures/lambdas, and async method + * control flow. + * + * <p>Await usage patterns: + * <ul> + * <li>Inside {@code @Async} methods: {@code await expr} (keyword syntax)</li> + * <li>Inside {@code async { }} closures: {@code await(expr)} with static import</li> + * <li>At script top-level: {@code AsyncUtils.await(expr)} or static import</li> + * </ul> + */ +class AsyncPatternsTest { + + @AfterEach + void resetExecutor() { + AsyncUtils.setExecutor(null) + } + + // ========================================================================= + // CPU-Bound async work + // ========================================================================= + + @Test + void testCpuBoundWithAsyncClosure() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def calculateComplexOutput(String input) { + new StringBuilder(input).reverse().toString() + } + + def calculateResultAsync(String input) { + def task = async { + def compute = async { calculateComplexOutput(input) } + def result = await(compute()) + return result + } + return task() + } + + assert await(calculateResultAsync("Hello, World!")) == "!dlroW ,olleH" + ''' + } + + @Test + void testCpuBoundWithAsyncAnnotation() { + assertScript ''' + import groovy.transform.Async + import static groovy.concurrent.AsyncUtils.* + + class ComputeService { + private String calculateComplexOutput(String input) { + new StringBuilder(input).reverse().toString() + } + + @Async + def calculateResult(String input) { + return calculateComplexOutput(input) + } + } + + def service = new ComputeService() + assert await(service.calculateResult("Groovy")) == "yvoorG" + ''' + } + + // ========================================================================= + // I/O-Bound async simulation + // ========================================================================= + + @Test + void testIoBoundSimulated() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync(String url) { + def fetch = async { + await(delay(50)) + return "Response from ${url}" + } + return fetch() + } + + def downloadDataAsync() { + def download = async { + def result = await(fetchDataAsync("https://example.com/data";)) + return result + } + return download() + } + + assert await(downloadDataAsync()) == "Response from https://example.com/data"; + ''' + } + + // ========================================================================= + // Caller perspectives + // ========================================================================= + + @Test + void testCallerAwaitImmediately() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { + await(delay(50)) + return "fetched data" + } + return fetch() + } + + assert await(fetchDataAsync()) == "fetched data" + ''' + } + + @Test + void testCallerStartThenAwaitLater() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { + await(delay(50)) + return "fetched data" + } + return fetch() + } + + def caller = async { + def task = fetchDataAsync() + def localResult = 1 + 2 + 3 + def downloadResult = await(task) + return "${downloadResult} (local: ${localResult})" + } + def result = await(caller()) + assert result == "fetched data (local: 6)" + ''' + } + + @Test + void testCallerReturnsTaskWithoutAwaiting() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { + await(delay(30)) + return "data" + } + return fetch() + } + + def caller() { + def task = fetchDataAsync() + return task + } + + assert await(caller()) == "data" + ''' + } + + // ========================================================================= + // Task continuations + // ========================================================================= + + @Test + void testTaskContinuationWithThen() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchDataAsync() { + def fetch = async { "Hello, World!" } + return fetch() + } + + def calculateComplexOutput(String input) { + new StringBuilder(input).reverse().toString() + } + + def result = await( + fetchDataAsync().then { data -> calculateComplexOutput(data) } + ) + assert result == "!dlroW ,olleH" + ''' + } + + @Test + void testTaskContinuationChaining() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def chain = async { 10 } + def result = await( + chain() + .then { it * 2 } + .then { it + 5 } + .then { "Result: $it" } + ) + assert result == "Result: 25" + ''' + } + + @Test + void testTaskContinuationWithThenCompose() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchUser(int id) { + def user = async { [id: id, name: "User${id}"] } + return user() + } + + def fetchOrders(int userId) { + def orders = async { ["order1", "order2"] } + return orders() + } + + def result = await( + fetchUser(42).thenCompose { user -> + fetchOrders(user.id).then { orders -> + [user: user, orders: orders] + } + } + ) + + assert result.user.name == "User42" + assert result.orders == ["order1", "order2"] + ''' + } + + // ========================================================================= + // Parallel execution: awaitAll / awaitAny + // ========================================================================= + + @Test + void testAwaitAllParallelTasks() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fetchPage(String url) { + def page = async { + await(delay(30)) + return "Content of ${url} (${url.length()} chars)" + } + return page() + } + + def results = awaitAll( + fetchPage("https://www.google.com/";), + fetchPage("https://www.microsoft.com/";), + fetchPage("https://www.example.com/";) + ) + + assert results.size() == 3 + assert results[0].contains("google") + assert results[1].contains("microsoft") + assert results[2].contains("example") + ''' + } + + @Test + void testAwaitAnyRace() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def fastTask = async { + await(delay(10)) + return "fast" + } + def fast = fastTask() + def slowTask = async { + await(delay(2000)) + return "slow" + } + def slow = slowTask() + + assert awaitAny(fast, slow) == "fast" + ''' + } + + // ========================================================================= + // Concurrent interleaving workflow + // ========================================================================= + + @Test + void testConcurrentTasksInterleaving() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def log = Collections.synchronizedList([]) + + def methodAAsync(List log) { + def method = async { + for (int i = 0; i < 5; i++) { + log << ("A" + i) + await(delay(50)) + } + log << "A returns result 123" + return 123 + } + return method() + } + + def taskA = methodAAsync(log) + + for (int i = 0; i < 5; i++) { + log << ("B" + i) + Thread.sleep(25) + } + + log << "Wait for taskA termination" + def result = await(taskA) + log << ("The result of taskA is " + result) + + assert result == 123 + assert log.contains("A0") + assert log.contains("B0") + assert log.contains("A returns result 123") + assert log.last() == "The result of taskA is 123" + ''' + } + + // ========================================================================= + // Delay patterns + // ========================================================================= + + @Test + void testAsyncDelayInLoop() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def loop = async { + def sum = 0 + for (int i = 1; i <= 5; i++) { + await(delay(10)) + sum += i + } + return sum + } + def result = await(loop()) + assert result == 15 + ''' + } + + @Test + void testDelayZeroCompletesImmediately() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + def task = async { + await(delay(0)) + return "done" + } + def result = await(task()) + assert result == "done" + ''' + } + + @Test + void testDelayNegativeThrows() { + assertScript ''' + import groovy.concurrent.AsyncUtils + + try { + AsyncUtils.delay(-1) + assert false : "Should have thrown" + } catch (IllegalArgumentException e) { + assert e.message.contains("negative") + } + ''' + } + + @Test + void testDelayWithTimeUnit() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + import java.util.concurrent.TimeUnit + + def task = async { + await(delay(50, TimeUnit.MILLISECONDS)) + return "delayed" + } + def result = await(task()) + assert result == "delayed" + ''' + } + + @Test + void testDelayIsNonBlocking() { + assertScript ''' + import static groovy.concurrent.AsyncUtils.* + + long startTime = System.currentTimeMillis() + def delayed = async { + await(delay(200)) + return "delayed" + } + def task = delayed() + + long afterStart = System.currentTimeMillis() + assert (afterStart - startTime) < 100 : "async should return immediately" + + def result = await(task) + long afterAwait = System.currentTimeMillis() + + assert result == "delayed" + assert (afterAwait - startTime) >= 180 : "should have waited for delay" Review Comment: This test relies on wall-clock timing (`< 100ms` to start and `>= 180ms` total) which is prone to CI flakiness (GC pauses, clock granularity, noisy neighbors). Consider asserting non-blocking behavior via deterministic signals (e.g., a latch that must be reachable before awaiting), or relax/remove the strict time thresholds. ```suggestion def delayed = async { await(delay(200)) return "delayed" } def task = delayed() def result = await(task) assert result == "delayed" ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
