Minor tweaks

Author: daniellansun

src/main/java/groovy/concurrent/AsyncScope.java

@@ -0,0 +1,256 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ */
+package groovy.concurrent;
+
+import groovy.lang.Closure;
+import org.apache.groovy.runtime.async.AsyncSupport;
+import org.apache.groovy.runtime.async.GroovyPromise;
+
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Structured concurrency scope that ensures all child tasks complete
+ * (or are cancelled) before the scope exits.
+ *
+ * <h2>Design philosophy</h2>
+ * <p>Inspired by Swift's {@code TaskGroup}, Kotlin's {@code coroutineScope},
+ * and JEP 453 (Structured Concurrency), {@code AsyncScope} provides a
+ * bounded lifetime for async tasks.  Unlike fire-and-forget
+ * {@link AsyncSupport#executeAsync}, tasks launched within a scope are
+ * guaranteed to complete before the scope closes.  This prevents:
+ * <ul>
+ *   <li>Orphaned tasks that outlive their logical parent</li>
+ *   <li>Resource leaks from uncollected async work</li>
+ *   <li>Silent failures from unobserved exceptions</li>
+ * </ul>
+ *
+ * <h2>Failure policy</h2>
+ * <p>By default, the scope uses a <b>fail-fast</b> policy: when any child
+ * task completes exceptionally, all sibling tasks are cancelled immediately.
+ * The first failure becomes the primary exception; subsequent failures are
+ * added as {@linkplain Throwable#addSuppressed(Throwable) suppressed}
+ * exceptions.  This matches Kotlin's
+ * {@code coroutineScope} / {@code supervisorScope} semantics.</p>
+ *
+ * <h2>Usage in Groovy</h2>
+ * <pre>
+ * import groovy.concurrent.AsyncScope
+ * import groovy.concurrent.Awaitable
+ *
+ * def results = AsyncScope.withScope { scope -&gt;
+ *     def userTask  = scope.async { fetchUser(id) }
+ *     def orderTask = scope.async { fetchOrders(id) }
+ *     return [user: await(userTask), orders: await(orderTask)]
+ * }
+ * // Both tasks guaranteed complete here
+ * </pre>
+ *
+ * <h2>Thread safety</h2>
+ * <p>All public methods are thread-safe.  The child task list uses
+ * {@link CopyOnWriteArrayList} for safe concurrent iteration during
+ * cancellation.  The {@link #closed} flag uses {@link AtomicBoolean}
+ * with CAS for exactly-once close semantics.</p>
+ *
+ * @see Awaitable
+ * @see AsyncSupport
+ * @since 6.0.0
+ */
+public class AsyncScope implements AutoCloseable {
+
+    private final List<CompletableFuture<?>> children = new CopyOnWriteArrayList<>();
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final Executor executor;
+    private final boolean failFast;
+
+    /**
+     * Creates a new scope with the given executor and fail-fast policy.
+     *
+     * @param executor the executor for child tasks; must not be {@code null}
+     * @param failFast if {@code true}, cancel all siblings when any child fails
+     */
+    public AsyncScope(Executor executor, boolean failFast) {
+        this.executor = executor;
+        this.failFast = failFast;
+    }
+
+    /**
+     * Creates a new scope with the given executor and fail-fast enabled.
+     *
+     * @param executor the executor for child tasks; must not be {@code null}
+     */
+    public AsyncScope(Executor executor) {
+        this(executor, true);
+    }
+
+    /**
+     * Creates a new scope with the default async executor and fail-fast enabled.
+     */
+    public AsyncScope() {
+        this(AsyncSupport.getExecutor(), true);
+    }
+
+    /**
+     * Launches a child task within this scope.  The task's lifetime is
+     * bound to the scope: when the scope is closed, all incomplete child
+     * tasks are cancelled.
+     *
+     * @param body the async body to execute
+     * @param <T>  the result type
+     * @return an {@link Awaitable} representing the child task
+     * @throws IllegalStateException if the scope has already been closed
+     */
+    @SuppressWarnings("unchecked")
+    public <T> Awaitable<T> async(Closure<T> body) {
+        if (closed.get()) {
+            throw new IllegalStateException("AsyncScope is closed — cannot launch new tasks");
+        }
+        CompletableFuture<T> cf = CompletableFuture.supplyAsync(() -> {
+            try {
+                return body.call();
+            } catch (CompletionException ce) {
+                throw ce;
+            } catch (Throwable t) {
+                throw new CompletionException(t);
+            }
+        }, executor);
+        children.add(cf);
+        if (failFast) {
+            cf.whenComplete((v, err) -> {
+                if (err != null && !closed.get()) {
+                    cancelAll();
+                }
+            });
+        }
+        return GroovyPromise.of(cf);
+    }
+
+    /**
+     * Returns the number of child tasks launched within this scope.
+     *
+     * @return the child task count
+     */
+    public int getChildCount() {
+        return children.size();
+    }
+
+    /**
+     * Cancels all child tasks.  Idempotent — safe to call multiple times.
+     * <p>
+     * Cancels each child via {@link CompletableFuture#cancel(boolean)}.
+     * Does <em>not</em> close the scope — the scope remains open so that
+     * {@link #close()} can still join all children and collect errors.
+     */
+    public void cancelAll() {
+        for (CompletableFuture<?> child : children) {
+            child.cancel(true);
+        }
+    }
+
+    /**
+     * Waits for all child tasks to complete, then closes the scope.
+     * <p>
+     * If any child failed, the first failure is rethrown with subsequent
+     * failures as {@linkplain Throwable#addSuppressed(Throwable) suppressed}
+     * exceptions.  Cancelled tasks are silently ignored.
+     * <p>
+     * This method is idempotent: only the first invocation waits for
+     * children; subsequent calls are no-ops.
+     */
+    @Override
+    public void close() {
+        if (!closed.compareAndSet(false, true)) return;
+        Throwable firstError = null;
+        for (CompletableFuture<?> child : children) {
+            try {
+                child.join();
+            } catch (CancellationException ignored) {
+                // Cancelled tasks are silently ignored
+            } catch (CompletionException e) {
+                Throwable cause = AsyncSupport.deepUnwrap(e);
+                if (cause instanceof CancellationException) {
+                    continue;
+                }
+                if (firstError == null) {
+                    firstError = cause;
+                } else {
+                    firstError.addSuppressed(cause);
+                }
+            } catch (Exception e) {
+                if (firstError == null) {
+                    firstError = e;
+                } else {
+                    firstError.addSuppressed(e);
+                }
+            }
+        }
+        if (firstError != null) {
+            if (firstError instanceof RuntimeException re) throw re;
+            if (firstError instanceof Error err) throw err;
+            throw new RuntimeException(firstError);
+        }
+    }
+
+    /**
+     * Convenience method that creates a scope, executes the given closure
+     * within it, and ensures the scope is closed on exit.
+     * <p>
+     * The closure receives the {@code AsyncScope} as its argument and can
+     * launch child tasks via {@link #async(Closure)}.  The scope is
+     * automatically closed (and all children awaited) when the closure
+     * returns or throws.
+     *
+     * <pre>
+     * def result = AsyncScope.withScope { scope -&gt;
+     *     def a = scope.async { computeA() }
+     *     def b = scope.async { computeB() }
+     *     return [await(a), await(b)]
+     * }
+     * </pre>
+     *
+     * @param body the closure to execute within the scope
+     * @param <T>  the result type
+     * @return the closure's return value
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> T withScope(Closure<T> body) {
+        return withScope(AsyncSupport.getExecutor(), body);
+    }
+
+    /**
+     * Convenience method that creates a scope with the given executor,
+     * executes the closure, and ensures the scope is closed on exit.
+     *
+     * @param executor the executor for child tasks
+     * @param body     the closure to execute within the scope
+     * @param <T>      the result type
+     * @return the closure's return value
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> T withScope(Executor executor, Closure<T> body) {
+        try (AsyncScope scope = new AsyncScope(executor)) {
+            return body.call(scope);
+        }
+    }
+}