You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@groovy.apache.org by "paulk-asert (via GitHub)" <gi...@apache.org> on 2026/04/15 12:49:35 UTC
[PR] GEB-18: Integrated Concurrency and Parallel Processing [groovy]
paulk-asert opened a new pull request, #2468:
URL: https://github.com/apache/groovy/pull/2468
(no comment)
--
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: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GROOVY-11953: Integrated Concurrency and Parallel Processing (publisher glue) [groovy]
Posted by "paulk-asert (via GitHub)" <gi...@apache.org>.
paulk-asert closed pull request #2468: GROOVY-11953: Integrated Concurrency and Parallel Processing (publisher glue)
URL: https://github.com/apache/groovy/pull/2468
--
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: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GEP-18: Integrated Concurrency and Parallel Processing [groovy]
Posted by "daniellansun (via GitHub)" <gi...@apache.org>.
daniellansun commented on PR #2468:
URL: https://github.com/apache/groovy/pull/2468#issuecomment-4272054271
+1
--
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: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GEP-18: Integrated Concurrency and Parallel Processing [groovy]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2468:
URL: https://github.com/apache/groovy/pull/2468#discussion_r3087665749
##########
src/main/java/groovy/concurrent/ChannelSelect.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent;
+
+import org.apache.groovy.runtime.async.AsyncSupport;
+import org.apache.groovy.runtime.async.GroovyPromise;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Selects the first available value from multiple {@link AsyncChannel}s.
+ * <p>
+ * This is the channel equivalent of {@link Awaitable#any(Object...)} —
+ * while {@code Awaitable.any} races futures, {@code ChannelSelect} races
+ * channel receives. Each call to {@link #select()} returns an
+ * {@link Awaitable} that completes with a {@link Result} indicating
+ * which channel produced the value and what it was.
+ *
+ * <pre>{@code
+ * def prices = AsyncChannel.create(10)
+ * def alerts = AsyncChannel.create(10)
+ *
+ * def sel = ChannelSelect.from(prices, alerts)
+ * def result = await sel.select()
+ * println "Channel ${result.index}: ${result.value}"
+ * }</pre>
+ * <p>
+ * Inspired by GPars' {@code Select} and Go's {@code select} statement.
+ *
+ * @since 6.0.0
+ */
+public final class ChannelSelect {
+
+ private final List<AsyncChannel<?>> channels;
+
+ private ChannelSelect(List<AsyncChannel<?>> channels) {
+ this.channels = channels;
+ }
+
+ /**
+ * Creates a select over the given channels.
+ *
+ * @param channels the channels to select from
+ * @return a new ChannelSelect
+ */
+ @SafeVarargs
+ public static ChannelSelect from(AsyncChannel<?>... channels) {
+ Objects.requireNonNull(channels, "channels must not be null");
+ if (channels.length == 0) {
+ throw new IllegalArgumentException("At least one channel is required");
+ }
+ return new ChannelSelect(List.of(channels));
+ }
+
+ /**
+ * Waits for the first value available from any of the channels.
+ * <p>
+ * Returns an {@link Awaitable} that completes with a {@link Result}
+ * containing the channel index and the received value.
+ *
+ * @return an awaitable result indicating which channel produced the value
+ */
+ public Awaitable<Result> select() {
+ CompletableFuture<Result> winner = new CompletableFuture<>();
+ for (int i = 0; i < channels.size(); i++) {
+ final int index = i;
+ AsyncChannel<?> ch = channels.get(i);
+ ch.receive().toCompletableFuture().whenComplete((value, error) -> {
+ if (error == null) {
+ winner.complete(new Result(index, value));
+ }
+ // On error (e.g., channel closed), let other channels compete
+ });
+ }
+ return GroovyPromise.of(winner);
+ }
Review Comment:
This implementation starts a real `receive()` on *every* channel. When a non-winning channel later receives a value, it will still consume that value but the result is discarded (because `winner.complete(...)` no-ops after the first completion). That can permanently lose messages and break subsequent selects/receives. A correct select needs non-consuming readiness registration, or a cancellable/transactional receive where losing registrations don't dequeue values. Consider adding/selecting via an AsyncChannel primitive that supports selection semantics (e.g., a `selectRegister` API) rather than racing `receive()` calls.
##########
src/main/java/groovy/concurrent/Agent.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent;
+
+import org.apache.groovy.runtime.async.AsyncSupport;
+import org.apache.groovy.runtime.async.GroovyPromise;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
+
+/**
+ * A thread-safe mutable-value container inspired by Clojure's agents and
+ * GPars' {@code Agent}.
+ * <p>
+ * An {@code Agent} wraps a value that can be read by any thread but
+ * modified only through serialised update functions. Updates are queued
+ * and applied one at a time on a dedicated executor, guaranteeing that
+ * the value is never corrupted by concurrent writes.
+ * <p>
+ * Reading the current value via {@link #get()} is non-blocking and
+ * returns a snapshot. Sending an update via {@link #send(Function)} is
+ * also non-blocking — the function is queued and applied asynchronously.
+ * Use {@link #sendAndGet(Function)} to obtain an {@link Awaitable} that
+ * completes with the new value after the update is applied.
+ *
+ * <pre>{@code
+ * // Groovy:
+ * def counter = Agent.create(0)
+ * counter.send { it + 1 }
+ * counter.send { it + 1 }
+ * assert await(counter.getAsync()) == 2
+ *
+ * // Java:
+ * Agent<Integer> counter = Agent.create(0);
+ * counter.send(n -> n + 1);
+ * Awaitable<Integer> result = counter.sendAndGet(n -> n + 1);
+ * }</pre>
+ *
+ * @param <T> the value type
+ * @see Pool
+ * @see AsyncScope
+ * @since 6.0.0
+ */
+public final class Agent<T> {
+
+ private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final ExecutorService updateExecutor;
+ private volatile T value;
+
+ private Agent(T initialValue, ExecutorService executor) {
+ this.value = initialValue;
+ this.updateExecutor = executor;
+ }
+
+ /**
+ * Creates an agent with the given initial value, using a
+ * single-thread executor for serialised updates.
+ *
+ * @param initialValue the starting value
+ * @param <T> the value type
+ * @return a new agent
+ */
+ public static <T> Agent<T> create(T initialValue) {
+ return new Agent<>(initialValue,
+ Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(r, "groovy-agent");
+ t.setDaemon(true);
+ return t;
+ }));
+ }
+
+ /**
+ * Creates an agent backed by the given pool for update execution.
+ * Updates are still serialised (only one at a time), but they run
+ * on the pool's threads.
+ *
+ * @param initialValue the starting value
+ * @param pool the pool to use for updates
+ * @param <T> the value type
+ * @return a new agent
+ */
+ public static <T> Agent<T> create(T initialValue, Pool pool) {
+ Objects.requireNonNull(pool, "pool must not be null");
+ // Wrap pool in a single-thread executor to serialise updates
+ return new Agent<>(initialValue,
+ Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(() -> {
+ pool.execute(r);
+ }, "groovy-agent-dispatch");
+ t.setDaemon(true);
+ return t;
+ }));
+ }
Review Comment:
This `ThreadFactory` implementation breaks `newSingleThreadExecutor`: the executor's worker thread is expected to run the executor's internal queue loop, but here it immediately submits that loop to `pool.execute(r)` and then exits. That can lead to a dead executor, unbounded thread creation, or tasks never executing. If you want updates to run on the provided `Pool` while staying serialized, implement a serializing executor (e.g., a `SerialExecutor` over `pool`) rather than delegating the worker thread runnable to another executor.
##########
src/main/java/org/apache/groovy/runtime/async/GeneratorBridge.java:
##########
@@ -150,7 +150,7 @@ public void close() {
/**
* Thrown when a generator tries to yield after the consumer has closed the bridge.
*/
- static final class GeneratorClosedException extends RuntimeException {
+ public static final class GeneratorClosedException extends RuntimeException {
Review Comment:
`complete()`, `completeExceptionally(...)`, and `GeneratorClosedException` were previously package-private/internal hooks (per the surrounding comments) and are now public API surface. If external callers aren't meant to control generator lifecycle, consider keeping these package-private and exposing only what’s necessary. Expanding public API makes future refactors harder and increases compatibility obligations.
##########
src/main/java/org/codehaus/groovy/runtime/ParallelCollectionExtensions.java:
##########
@@ -0,0 +1,275 @@
+/*
+ * 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.runtime;
+
+import groovy.concurrent.Pool;
+import org.codehaus.groovy.runtime.InvokerHelper;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ForkJoinPool;
+import java.util.function.BiConsumer;
+import java.util.function.BinaryOperator;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * DGM-like extension methods that add parallel collection operations
+ * to {@link Collection}.
+ * <p>
+ * These methods use {@link Pool#current()} to obtain the current pool
+ * (typically set by {@link groovy.concurrent.ParallelScope#withPool}).
+ * If no pool is current, they fall back to {@link ForkJoinPool#commonPool()}.
+ * <p>
+ * All methods delegate to Java parallel streams with pool isolation:
+ * the stream operations run on the current pool's {@link ForkJoinPool},
+ * not the common pool.
+ * <p>
+ * Inspired by GPars' {@code GParsPoolUtil} category methods.
+ *
+ * @since 6.0.0
+ */
+public class ParallelCollectionExtensions {
+
+ // ---- Iteration ------------------------------------------------------
+
+ /**
+ * Iterates over the collection in parallel, applying the action
+ * to each element.
+ */
+ public static <T> void eachParallel(Collection<T> self, Consumer<T> action) {
+ withCurrentFJP(fjp ->
+ fjp.submit(() -> self.parallelStream().forEach(action)).join()
+ );
+ }
+
+ // ---- Transformation -------------------------------------------------
+
+ /**
+ * Transforms each element in parallel, returning a new list.
+ */
+ public static <T, R> List<R> collectParallel(Collection<T> self, Function<T, R> transform) {
+ return withCurrentFJP(fjp ->
+ fjp.submit(() -> self.parallelStream().map(transform).collect(Collectors.toList())).join()
+ );
+ }
+
+ // ---- Filtering ------------------------------------------------------
+
+ /**
+ * Filters the collection in parallel, returning elements that match.
+ */
+ public static <T> List<T> findAllParallel(Collection<T> self, Predicate<T> filter) {
+ return withCurrentFJP(fjp ->
+ fjp.submit(() -> self.parallelStream().filter(filter).collect(Collectors.toList())).join()
+ );
+ }
+
+ /**
+ * Finds the first element matching the predicate (order not guaranteed
+ * due to parallelism).
+ */
+ public static <T> T findParallel(Collection<T> self, Predicate<T> filter) {
+ return withCurrentFJP(fjp ->
+ fjp.submit(() -> self.parallelStream().filter(filter).findFirst().orElse(null)).join()
+ );
+ }
Review Comment:
The doc says order is not guaranteed, but `findFirst()` on an ordered stream *does* respect encounter order (and may incur extra coordination cost). If the intention is 'no order guarantee', this should use `findAny()` instead; otherwise, update the Javadoc to reflect that it returns the first element in encounter order for ordered collections.
##########
src/main/java/groovy/concurrent/ConcurrentConfig.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent;
+
+import org.apache.groovy.runtime.async.AsyncSupport;
+
+/**
+ * Central configuration for Groovy's concurrent and parallel features.
+ * <p>
+ * Configuration is resolved in order:
+ * <ol>
+ * <li>Programmatic overrides via setter methods</li>
+ * <li>System properties ({@code groovy.concurrent.poolsize},
+ * {@code groovy.concurrent.virtual})</li>
+ * <li>Sensible defaults (processors + 1, virtual threads on JDK 21+)</li>
+ * </ol>
+ * <p>
+ * Inspired by GPars' {@code GParsConfig} and {@code PoolUtils}, modernised
+ * for virtual threads and the {@link Pool} abstraction.
+ *
+ * @see Pool
+ * @since 6.0.0
+ */
+public final class ConcurrentConfig {
+
+ private static volatile int poolSizeOverride = 0; // 0 means "not set"
+ private static volatile Pool defaultPool;
+
+ private ConcurrentConfig() { }
+
+ /**
+ * Returns the default parallelism level.
+ * <p>
+ * Checks (in order): programmatic override, system property
+ * {@code groovy.concurrent.poolsize}, then
+ * {@code Runtime.availableProcessors() + 1} (following the GPars convention).
+ *
+ * @return the default pool size, always > 0
+ */
+ public static int getDefaultParallelism() {
+ if (poolSizeOverride > 0) return poolSizeOverride;
+ try {
+ String prop = System.getProperty("groovy.concurrent.poolsize");
+ if (prop != null) {
+ int val = Integer.parseInt(prop);
+ if (val > 0) return val;
+ }
+ } catch (SecurityException | NumberFormatException ignored) { }
+ return Runtime.getRuntime().availableProcessors() + 1;
+ }
+
+ /**
+ * Sets the default parallelism level programmatically.
+ * Pass {@code 0} or a negative value to reset to the default.
+ *
+ * @param size the pool size, or ≤ 0 to reset
+ */
+ public static void setDefaultParallelism(int size) {
+ poolSizeOverride = Math.max(0, size);
+ }
+
+ /**
+ * Returns {@code true} if virtual threads should be preferred.
+ * <p>
+ * Checks the system property {@code groovy.concurrent.virtual}
+ * (default: {@code true} on JDK 21+, {@code false} otherwise).
+ *
+ * @return whether to prefer virtual threads
+ */
+ public static boolean preferVirtualThreads() {
+ try {
+ String prop = System.getProperty("groovy.concurrent.virtual");
+ if (prop != null) return Boolean.parseBoolean(prop);
+ } catch (SecurityException ignored) { }
+ return AsyncSupport.isVirtualThreadsAvailable();
+ }
+
+ /**
+ * Returns the global default pool. If none has been set, creates one
+ * based on current configuration: a virtual-thread pool if
+ * {@link #preferVirtualThreads()} is {@code true}, otherwise a
+ * fixed pool sized by {@link #getDefaultParallelism()}.
+ *
+ * @return the default pool, never {@code null}
+ */
+ public static Pool getDefaultPool() {
+ Pool p = defaultPool;
+ if (p != null) return p;
+ return preferVirtualThreads() ? Pool.virtual() : Pool.fixed(getDefaultParallelism());
Review Comment:
`getDefaultPool()` claims to return a *global default pool* but currently creates and returns a new pool instance on every call unless `defaultPool` was explicitly set. This will leak threads/executors and can cause surprising behavior. Consider lazily initializing and caching `defaultPool` (e.g., double-checked locking or synchronized init) and document/reset its lifecycle clearly.
```suggestion
* Returns the global default pool. If none has been set, lazily creates
* and caches one based on current configuration: a virtual-thread pool if
* {@link #preferVirtualThreads()} is {@code true}, otherwise a
* fixed pool sized by {@link #getDefaultParallelism()}.
* <p>
* The created pool is reused for subsequent calls until replaced or reset
* via {@link #setDefaultPool(Pool)}.
*
* @return the default pool, never {@code null}
*/
public static Pool getDefaultPool() {
Pool p = defaultPool;
if (p != null) return p;
synchronized (ConcurrentConfig.class) {
p = defaultPool;
if (p == null) {
p = preferVirtualThreads() ? Pool.virtual() : Pool.fixed(getDefaultParallelism());
defaultPool = p;
}
}
return p;
```
##########
src/main/java/groovy/concurrent/AsyncScope.java:
##########
@@ -115,35 +121,94 @@ static <T> T withCurrent(AsyncScope scope, Supplier<T> supplier) {
}
/**
- * Creates a scope, executes the closure within it, and ensures the
- * scope is closed on exit. The closure receives the scope as its
+ * Creates a scope, executes the body within it, and ensures the
+ * scope is closed on exit. The body receives the scope as its
* argument for launching child tasks.
*
* <pre>{@code
+ * // Java:
+ * var result = AsyncScope.withScope(scope -> {
+ * var a = scope.async(() -> computeA());
+ * var b = scope.async(() -> computeB());
+ * return List.of(AsyncSupport.await(a), AsyncSupport.await(b));
+ * });
+ *
+ * // Groovy (Closure overload added via extension method):
* def result = AsyncScope.withScope { scope ->
* def a = scope.async { computeA() }
* def b = scope.async { computeB() }
* return [await(a), await(b)]
* }
* }</pre>
+ *
+ * @param body the function to execute within the scope
+ * @param <T> the result type
+ * @return the body's return value
*/
- @SuppressWarnings("unchecked")
- static <T> T withScope(
- @ClosureParams(value = SimpleType.class, options = "groovy.concurrent.AsyncScope") Closure<T> body) {
+ static <T> T withScope(Function<AsyncScope, T> body) {
return withScope(AsyncSupport.getExecutor(), body);
}
/**
- * Creates a scope with the given executor, executes the closure,
+ * Creates a scope with the given executor, executes the body,
* and ensures the scope is closed on exit.
+ *
+ * @param executor the executor for child tasks
+ * @param body the function to execute within the scope
+ * @param <T> the result type
+ * @return the body's return value
*/
- @SuppressWarnings("unchecked")
- static <T> T withScope(Executor executor,
- @ClosureParams(value = SimpleType.class, options = "groovy.concurrent.AsyncScope") Closure<T> body) {
+ static <T> T withScope(Executor executor, Function<AsyncScope, T> body) {
Objects.requireNonNull(body, "body must not be null");
- try (AsyncScope scope = create(executor)) {
- return withCurrent(scope, () -> body.call(scope));
+ AsyncScope scope = create(executor);
+ T result;
+ try {
+ result = withCurrent(scope, () -> body.apply(scope));
+ } catch (Throwable bodyError) {
+ try {
+ scope.close();
+ } catch (Throwable closeError) {
+ if (closeError != bodyError) {
+ bodyError.addSuppressed(closeError);
+ }
+ }
+ if (bodyError instanceof RuntimeException re) throw re;
+ if (bodyError instanceof Error err) throw err;
+ throw new RuntimeException(bodyError);
}
+ scope.close();
+ return result;
+ }
+
+ /**
+ * Creates a scope with a timeout. If the body does not complete within
+ * the specified duration, all child tasks are cancelled and the scope
+ * throws {@link java.util.concurrent.TimeoutException}.
+ *
+ * @param timeout the maximum duration for the scope
+ * @param body the function to execute within the scope
+ * @param <T> the result type
+ * @return the body's return value
+ * @throws java.util.concurrent.TimeoutException if the timeout expires
+ * @since 6.0.0
+ */
Review Comment:
The API docs here specify `TimeoutException`, but `DefaultAsyncScope.withScopeTimeout(...)` currently throws `CancellationException` on timeout (and tests expect `CancellationException`). The Javadoc and the actual thrown type should be aligned (either update the implementation to throw `TimeoutException`, or update these docs to reflect `CancellationException`).
##########
src/main/java/org/apache/groovy/runtime/async/GeneratorBridge.java:
##########
@@ -71,7 +71,7 @@ public void yield(Object value) {
/**
* Called internally when the generator completes normally.
*/
- void complete() {
+ public void complete() {
Review Comment:
`complete()`, `completeExceptionally(...)`, and `GeneratorClosedException` were previously package-private/internal hooks (per the surrounding comments) and are now public API surface. If external callers aren't meant to control generator lifecycle, consider keeping these package-private and exposing only what’s necessary. Expanding public API makes future refactors harder and increases compatibility obligations.
##########
src/main/java/org/apache/groovy/runtime/async/GeneratorBridge.java:
##########
@@ -85,7 +85,7 @@ void complete() {
/**
* Called internally when the generator throws an exception.
*/
- void completeExceptionally(Throwable error) {
+ public void completeExceptionally(Throwable error) {
Review Comment:
`complete()`, `completeExceptionally(...)`, and `GeneratorClosedException` were previously package-private/internal hooks (per the surrounding comments) and are now public API surface. If external callers aren't meant to control generator lifecycle, consider keeping these package-private and exposing only what’s necessary. Expanding public API makes future refactors harder and increases compatibility obligations.
##########
src/main/java/groovy/concurrent/DataflowVariable.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent;
+
+import org.apache.groovy.runtime.async.GroovyPromise;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * A single-assignment variable for dataflow-style programming.
+ * <p>
+ * A {@code DataflowVariable} starts unbound. It can be written to exactly
+ * once via {@link #bind(Object)} (or the {@code <<} operator in Groovy).
+ * Any thread that reads the variable before it is bound will block until
+ * a value becomes available. Once bound, all subsequent reads return the
+ * same value immediately.
+ * <p>
+ * {@code DataflowVariable} implements {@link Awaitable}, so it works
+ * naturally with {@code await}:
+ *
+ * <pre>{@code
+ * def x = new DataflowVariable()
+ * def y = new DataflowVariable()
+ *
+ * def z = Awaitable.go { await(x) + await(y) }
+ *
+ * async { x << 10 }
+ * async { y << 5 }
+ *
+ * println "Result: ${await(z)}" // 15
+ * }</pre>
+ * <p>
+ * Inspired by GPars' {@code DataflowVariable}, modernised to integrate
+ * with Groovy's {@code async}/{@code await} and {@link Awaitable} API.
+ *
+ * @param <T> the value type
+ * @see Awaitable
+ * @see Dataflows
+ * @since 6.0.0
+ */
+public class DataflowVariable<T> implements Awaitable<T> {
+
+ private final CompletableFuture<T> future = new CompletableFuture<>();
+ private final Awaitable<T> awaitable = GroovyPromise.of(future);
+
+ /**
+ * Creates an unbound dataflow variable.
+ */
+ public DataflowVariable() { }
+
+ /**
+ * Binds this variable to the given value. Can only be called once;
+ * subsequent calls throw {@link IllegalStateException}.
+ *
+ * @param value the value to bind (may be {@code null})
+ * @throws IllegalStateException if already bound
+ */
+ public void bind(T value) {
+ if (!future.complete(value)) {
+ throw new IllegalStateException(
+ "DataflowVariable is already bound to: " + future.join());
+ }
+ }
Review Comment:
If the variable was previously completed exceptionally, `future.join()` will throw `CompletionException` while building the error message, so `bind()` will throw the wrong exception type (and potentially hide the intended IllegalStateException). Consider using `future.getNow(null)` (guarded), `future.isCompletedExceptionally()`, or a safe string representation that can't throw.
##########
src/main/java/groovy/concurrent/BroadcastChannel.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 groovy.concurrent;
+
+import org.apache.groovy.runtime.async.AsyncSupport;
Review Comment:
This import is unused in the current file and may trigger static analysis failures in builds with strict checks. Please remove it.
```suggestion
```
##########
src/main/java/org/codehaus/groovy/transform/ParallelASTTransformation.java:
##########
@@ -20,93 +20,126 @@
import groovy.transform.Parallel;
import org.apache.groovy.lang.annotation.Incubating;
-import org.apache.groovy.util.concurrent.ThreadHelper;
import org.codehaus.groovy.ast.ASTNode;
import org.codehaus.groovy.ast.AnnotationNode;
import org.codehaus.groovy.ast.ClassHelper;
+import org.codehaus.groovy.ast.ClassNode;
+import org.codehaus.groovy.ast.ClassCodeVisitorSupport;
+import org.codehaus.groovy.ast.ModuleNode;
import org.codehaus.groovy.ast.Parameter;
import org.codehaus.groovy.ast.VariableScope;
import org.codehaus.groovy.ast.expr.ClosureExpression;
+import org.codehaus.groovy.ast.expr.ConstantExpression;
import org.codehaus.groovy.ast.expr.Expression;
-import org.codehaus.groovy.ast.expr.VariableExpression;
+import org.codehaus.groovy.ast.expr.ListExpression;
import org.codehaus.groovy.ast.stmt.BlockStatement;
import org.codehaus.groovy.ast.stmt.ForStatement;
import org.codehaus.groovy.ast.stmt.Statement;
+import org.codehaus.groovy.classgen.GeneratorContext;
+import org.codehaus.groovy.control.CompilationFailedException;
import org.codehaus.groovy.control.CompilePhase;
import org.codehaus.groovy.control.SourceUnit;
-import org.codehaus.groovy.control.messages.SyntaxErrorMessage;
-import org.codehaus.groovy.syntax.SyntaxException;
+import org.codehaus.groovy.control.customizers.CompilationCustomizer;
+
+import java.util.ArrayList;
+import java.util.List;
import static org.codehaus.groovy.ast.tools.GeneralUtils.args;
import static org.codehaus.groovy.ast.tools.GeneralUtils.block;
import static org.codehaus.groovy.ast.tools.GeneralUtils.callX;
-import static org.codehaus.groovy.ast.tools.GeneralUtils.castX;
import static org.codehaus.groovy.ast.tools.GeneralUtils.stmt;
/**
- * Simple AST transform for {@link Parallel}: each {@code for-in} iteration body is
- * wrapped in a new thread and started immediately.
+ * Global AST transform for {@link Parallel}: rewrites annotated
+ * {@code for-in} loops into {@code collection.eachParallel { item -> body }}.
+ * <p>
+ * Runs at {@link CompilePhase#CONVERSION} — before variable scope
+ * resolution — so the compiler processes the generated closure normally,
+ * including proper capture of enclosing local variables.
*/
@Incubating
-@GroovyASTTransformation(phase = CompilePhase.SEMANTIC_ANALYSIS)
+@GroovyASTTransformation(phase = CompilePhase.CONVERSION)
public class ParallelASTTransformation implements ASTTransformation {
+ private static final ClassNode PARALLEL_ANNOTATION = ClassHelper.make(Parallel.class);
+
@Override
- public void visit(final ASTNode[] nodes, final SourceUnit source) {
- if (nodes == null || nodes.length != 2) return;
- if (!(nodes[0] instanceof AnnotationNode annotation)) return;
- if (!Parallel.class.getName().equals(annotation.getClassNode().getName())) return;
-
- if (!(nodes[1] instanceof ForStatement forStatement)) {
- source.getErrorCollector().addError(new SyntaxErrorMessage(
- new SyntaxException("@Parallel may only be applied to a for loop statement",
- annotation.getLineNumber(), annotation.getColumnNumber()), source));
- return;
- }
+ public void visit(ASTNode[] nodes, SourceUnit source) {
+ ModuleNode module = source.getAST();
+ if (module == null) return;
- if (forStatement.getValueVariable() == null) {
- source.getErrorCollector().addError(new SyntaxErrorMessage(
- new SyntaxException("@Parallel currently supports only for-in loops",
- annotation.getLineNumber(), annotation.getColumnNumber()), source));
- return;
+ for (ClassNode classNode : module.getClasses()) {
+ new ParallelForVisitor(source).visitClass(classNode);
}
-
- injectParallelThreadStart(forStatement, annotation);
}
- private static void injectParallelThreadStart(final ForStatement forStatement, final AnnotationNode annotation) {
- Statement originalBody = forStatement.getLoopBlock();
-
- // Use a closure parameter with the same loop variable name and curry(currentValue)
- // so each launched thread receives its own iteration value.
- Parameter loopParameter = forStatement.getValueVariable();
- Parameter workerParameter = new Parameter(loopParameter.getOriginType(), loopParameter.getName());
+ /**
+ * Visits all ForStatements looking for @Parallel annotations.
+ * When found, replaces the for loop body with a call to
+ * collection.eachParallel { item -> body }.
+ */
+ private static class ParallelForVisitor extends ClassCodeVisitorSupport {
+ private final SourceUnit source;
- BlockStatement workerCode = block();
- if (originalBody instanceof BlockStatement bodyBlock) {
- bodyBlock.getStatements().forEach(workerCode::addStatement);
- } else {
- workerCode.addStatement(originalBody);
+ ParallelForVisitor(SourceUnit source) {
+ this.source = source;
}
- ClosureExpression worker = new ClosureExpression(new Parameter[]{workerParameter}, workerCode);
- worker.setVariableScope(new VariableScope());
- worker.setSourcePosition(annotation);
+ @Override
+ protected SourceUnit getSourceUnit() {
+ return source;
+ }
- Expression currentLoopValue = new VariableExpression(loopParameter);
- Expression boundWorker = callX(worker, "curry", args(currentLoopValue));
- Expression runnable = castX(ClassHelper.make(Runnable.class), boundWorker);
+ @Override
+ public void visitForLoop(ForStatement forLoop) {
+ AnnotationNode parallelAnnotation = findParallelAnnotation(forLoop);
+ if (parallelAnnotation != null) {
+ rewriteForLoop(forLoop, parallelAnnotation);
+ }
+ super.visitForLoop(forLoop);
+ }
- Statement startThread = stmt(callX(
- ClassHelper.make(ThreadHelper.class),
- "startThread",
- args(runnable)));
- startThread.setSourcePosition(annotation);
+ private AnnotationNode findParallelAnnotation(ForStatement forLoop) {
+ for (AnnotationNode ann : forLoop.getStatementAnnotations()) {
+ if (ann.getClassNode().getName().equals(Parallel.class.getName())) {
+ return ann;
+ }
+ }
+ return null;
+ }
- BlockStatement loopBody = block(startThread);
- loopBody.setSourcePosition(originalBody);
- forStatement.setLoopBlock(loopBody);
+ private void rewriteForLoop(ForStatement forLoop, AnnotationNode annotation) {
+ Statement originalBody = forLoop.getLoopBlock();
+ Parameter loopParameter = forLoop.getValueVariable();
+ Expression originalCollection = forLoop.getCollectionExpression();
+
+ // Build closure: { item -> originalBody }
+ Parameter closureParam = new Parameter(
+ loopParameter.getOriginType(), loopParameter.getName());
+ BlockStatement closureBody = block();
+ if (originalBody instanceof BlockStatement bodyBlock) {
+ bodyBlock.getStatements().forEach(closureBody::addStatement);
+ } else {
+ closureBody.addStatement(originalBody);
+ }
+ ClosureExpression closure = new ClosureExpression(
+ new Parameter[]{closureParam}, closureBody);
+ closure.setVariableScope(new VariableScope());
+ closure.setSourcePosition(annotation);
+
+ // collection.eachParallel(closure)
+ Expression eachParallelCall = callX(
+ originalCollection, "eachParallel", args(closure));
+
+ Statement callStatement = stmt(eachParallelCall);
+ callStatement.setSourcePosition(annotation);
+
+ // Replace the for loop: iterate over [null] once,
+ // executing the eachParallel call
+ ListExpression singleRun = new ListExpression();
+ singleRun.addExpression(ConstantExpression.NULL);
+ forLoop.setCollectionExpression(singleRun);
+ forLoop.setLoopBlock(block(callStatement));
+ }
Review Comment:
Rewriting the annotated loop into a single-iteration `for (x in [null]) { collection.eachParallel { ... } }` changes observable semantics of `for` in Groovy (e.g., the loop variable may end up as `null` after the loop rather than the last iterated value; statement labeling and debugging/line mapping can also become confusing). Prefer replacing the parent statement with a direct expression/statement (`collection.eachParallel { ... }`) instead of mutating the loop to 'run once'. If replacing in-place is hard with `ClassCodeVisitorSupport`, consider a `CompilationCustomizer` that rewrites the enclosing `BlockStatement` statement list.
##########
src/main/java/groovy/concurrent/ChannelSelect.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent;
+
+import org.apache.groovy.runtime.async.AsyncSupport;
Review Comment:
This import is unused in the current file and may trigger static analysis failures in builds with strict checks. Please remove it.
```suggestion
```
--
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: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GEP-18: Integrated Concurrency and Parallel Processing [groovy]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2468:
URL: https://github.com/apache/groovy/pull/2468#discussion_r3094255556
##########
src/main/java/groovy/concurrent/Agent.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent;
+
+import org.apache.groovy.runtime.async.GroovyPromise;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
+
+/**
+ * A thread-safe mutable-value container inspired by Clojure's agents and
+ * GPars' {@code Agent}.
+ * <p>
+ * An {@code Agent} wraps a value that can be read by any thread but
+ * modified only through serialised update functions. Updates are queued
+ * and applied one at a time on a dedicated executor, guaranteeing that
+ * the value is never corrupted by concurrent writes.
+ * <p>
+ * Reading the current value via {@link #get()} is non-blocking and
+ * returns a snapshot. Sending an update via {@link #send(Function)} is
+ * also non-blocking — the function is queued and applied asynchronously.
+ * Use {@link #sendAndGet(Function)} to obtain an {@link Awaitable} that
+ * completes with the new value after the update is applied.
+ *
+ * <pre>{@code
+ * // Groovy:
+ * def counter = Agent.create(0)
+ * counter.send { it + 1 }
+ * counter.send { it + 1 }
+ * assert await(counter.getAsync()) == 2
+ *
+ * // Java:
+ * Agent<Integer> counter = Agent.create(0);
+ * counter.send(n -> n + 1);
+ * Awaitable<Integer> result = counter.sendAndGet(n -> n + 1);
+ * }</pre>
+ *
+ * @param <T> the value type
+ * @see Pool
+ * @see AsyncScope
+ * @since 6.0.0
+ */
+public final class Agent<T> {
+
+ private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final ExecutorService updateExecutor;
+ private volatile T value;
+
+ private Agent(T initialValue, ExecutorService executor) {
+ this.value = initialValue;
+ this.updateExecutor = executor;
+ }
+
+ /**
+ * Creates an agent with the given initial value, using a
+ * single-thread executor for serialised updates.
+ *
+ * @param initialValue the starting value
+ * @param <T> the value type
+ * @return a new agent
+ */
+ public static <T> Agent<T> create(T initialValue) {
+ return new Agent<>(initialValue,
+ Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(r, "groovy-agent");
+ t.setDaemon(true);
+ return t;
+ }));
+ }
+
+ /**
+ * Creates an agent backed by the given pool for update execution.
+ * Updates are still serialised (only one at a time), but they run
+ * on the pool's threads.
+ *
+ * @param initialValue the starting value
+ * @param pool the pool to use for updates
+ * @param <T> the value type
+ * @return a new agent
+ */
+ public static <T> Agent<T> create(T initialValue, Pool pool) {
+ Objects.requireNonNull(pool, "pool must not be null");
+ // Use a SerialExecutor to serialise updates on the pool's threads.
+ // We cannot use newSingleThreadExecutor with a delegating ThreadFactory
+ // because that breaks the executor's internal task loop.
+ return new Agent<>(initialValue, new SerialExecutor(pool));
+ }
+
+ /**
+ * Returns the current value. This is a non-blocking snapshot read.
+ *
+ * @return the current value
+ */
+ public T get() {
+ rwLock.readLock().lock();
+ try {
+ return value;
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ }
+
+ /**
+ * Returns the current value as an {@link Awaitable}. The awaitable
+ * completes after all previously queued updates have been applied,
+ * ensuring a consistent read.
+ *
+ * @return an awaitable holding the value after pending updates
+ */
+ public Awaitable<T> getAsync() {
+ CompletableFuture<T> cf = new CompletableFuture<>();
+ updateExecutor.execute(() -> {
+ rwLock.readLock().lock();
+ try {
+ cf.complete(value);
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ });
+ return GroovyPromise.of(cf);
+ }
+
+ /**
+ * Queues an update function to be applied to the current value.
+ * The function receives the current value and returns the new value.
+ * <p>
+ * Updates are applied asynchronously and serialised: only one update
+ * runs at a time.
+ *
+ * @param updateFn a function from current value to new value
+ */
+ public void send(Function<T, T> updateFn) {
+ Objects.requireNonNull(updateFn, "update function must not be null");
+ updateExecutor.execute(() -> applyUpdate(updateFn));
+ }
+
+ /**
+ * Queues an update function and returns an {@link Awaitable} that
+ * completes with the new value after the update is applied.
+ *
+ * @param updateFn a function from current value to new value
+ * @return an awaitable holding the new value
+ */
+ public Awaitable<T> sendAndGet(Function<T, T> updateFn) {
+ Objects.requireNonNull(updateFn, "update function must not be null");
+ CompletableFuture<T> cf = new CompletableFuture<>();
+ updateExecutor.execute(() -> {
+ try {
+ T newVal = applyUpdate(updateFn);
+ cf.complete(newVal);
+ } catch (Throwable t) {
+ cf.completeExceptionally(t);
+ }
+ });
+ return GroovyPromise.of(cf);
+ }
+
+ /**
+ * Shuts down the agent's update executor. No further updates will
+ * be accepted. Pending updates are executed before shutdown completes.
+ */
+ public void shutdown() {
+ updateExecutor.shutdown();
+ }
Review Comment:
The `shutdown()` Javadoc claims “Pending updates are executed before shutdown completes”, but `ExecutorService.shutdown()` is non-blocking and returns immediately (and the `SerialExecutor.shutdown()` implementation only flips a flag). Either update the documentation to avoid implying a synchronous drain, or provide an explicit “await termination/drain” API if that guarantee is important.
##########
src/test/groovy/groovy/concurrent/ChannelCompositionTest.groovy:
##########
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent
+
+import org.junit.jupiter.api.Test
+
+import static groovy.test.GroovyAssert.assertScript
+
+final class ChannelCompositionTest {
+
+ @Test
+ void testFilter() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+
+ def source = AsyncChannel.create(10)
+ def evens = source.filter { it % 2 == 0 }
+
+ async {
+ (1..10).each { source.send(it) }
+ source.close()
+ }
+
+ def results = []
+ for (val in evens) { results << val }
+ assert results == [2, 4, 6, 8, 10]
+ '''
+ }
+
+ @Test
+ void testMap() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+
+ def source = AsyncChannel.create(10)
+ def doubled = source.map { it * 2 }
+
+ async {
+ (1..5).each { source.send(it) }
+ source.close()
+ }
+
+ def results = []
+ for (val in doubled) { results << val }
+ assert results == [2, 4, 6, 8, 10]
+ '''
+ }
+
+ @Test
+ void testFilterAndMap() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+
+ def source = AsyncChannel.create(10)
+ def pipeline = source
+ .filter { it > 3 }
+ .map { it * 10 }
+
+ async {
+ (1..5).each { source.send(it) }
+ source.close()
+ }
+
+ def results = []
+ for (val in pipeline) { results << val }
+ assert results == [40, 50]
+ '''
+ }
+
+ @Test
+ void testMerge() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+
+ def ch1 = AsyncChannel.create(10)
+ def ch2 = AsyncChannel.create(10)
+ def merged = ch1.merge(ch2)
+
+ async {
+ ch1.send(1); ch1.send(2); ch1.close()
+ }
+ async {
+ ch2.send(10); ch2.send(20); ch2.close()
+ }
+
+ def results = []
+ for (val in merged) { results << val }
+ assert results.sort() == [1, 2, 10, 20]
+ '''
+ }
+
+ @Test @org.junit.jupiter.api.Disabled("split cleanup issue when run with other tests")
+ void testSplit() {
Review Comment:
This test is currently disabled with a note about a “split cleanup issue when run with other tests”. Leaving it disabled will hide regressions and suggests the underlying resource/closure semantics aren’t stable yet. Please either fix the cleanup issue so the test can run reliably, or remove/replace the test until the feature behavior is stable.
##########
src/main/java/org/codehaus/groovy/transform/ParallelASTTransformation.java:
##########
@@ -20,93 +20,126 @@
import groovy.transform.Parallel;
import org.apache.groovy.lang.annotation.Incubating;
-import org.apache.groovy.util.concurrent.ThreadHelper;
import org.codehaus.groovy.ast.ASTNode;
import org.codehaus.groovy.ast.AnnotationNode;
import org.codehaus.groovy.ast.ClassHelper;
+import org.codehaus.groovy.ast.ClassNode;
+import org.codehaus.groovy.ast.ClassCodeVisitorSupport;
+import org.codehaus.groovy.ast.ModuleNode;
import org.codehaus.groovy.ast.Parameter;
import org.codehaus.groovy.ast.VariableScope;
import org.codehaus.groovy.ast.expr.ClosureExpression;
+import org.codehaus.groovy.ast.expr.ConstantExpression;
import org.codehaus.groovy.ast.expr.Expression;
-import org.codehaus.groovy.ast.expr.VariableExpression;
+import org.codehaus.groovy.ast.expr.ListExpression;
import org.codehaus.groovy.ast.stmt.BlockStatement;
import org.codehaus.groovy.ast.stmt.ForStatement;
import org.codehaus.groovy.ast.stmt.Statement;
+import org.codehaus.groovy.classgen.GeneratorContext;
+import org.codehaus.groovy.control.CompilationFailedException;
import org.codehaus.groovy.control.CompilePhase;
import org.codehaus.groovy.control.SourceUnit;
-import org.codehaus.groovy.control.messages.SyntaxErrorMessage;
-import org.codehaus.groovy.syntax.SyntaxException;
+import org.codehaus.groovy.control.customizers.CompilationCustomizer;
+
+import java.util.ArrayList;
+import java.util.List;
Review Comment:
This file currently won’t compile: several imports are unused (e.g., GeneratorContext, CompilationFailedException, CompilationCustomizer, ArrayList, List). Java treats unused imports as compilation errors, so these must be removed (or the types actually used) before merge.
##########
src/main/java/org/apache/groovy/runtime/async/DefaultActor.java:
##########
@@ -0,0 +1,167 @@
+/*
+ * 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.Actor;
+import groovy.concurrent.Awaitable;
+import groovy.concurrent.DataflowVariable;
+
+import java.util.Objects;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+/**
+ * Default implementation of {@link Actor} using a dedicated thread
+ * and a {@link LinkedBlockingQueue} for message processing.
+ * <p>
+ * Each actor runs on its own thread (virtual on JDK 21+). Messages
+ * are processed one at a time, guaranteeing thread-safe state access
+ * without locks.
+ *
+ * @param <T> the message type
+ * @see Actor
+ * @since 6.0.0
+ */
+public final class DefaultActor<T> implements Actor<T> {
+
+ private static final Object POISON = new Object();
+
+ private final LinkedBlockingQueue<Envelope<T>> queue = new LinkedBlockingQueue<>();
+ private final MessageProcessor<T> processor;
+ private volatile boolean active = true;
+
+ private DefaultActor(MessageProcessor<T> processor) {
+ this.processor = processor;
+ AsyncSupport.getExecutor().execute(this::processLoop);
+ }
+
Review Comment:
`DefaultActor` starts its long-lived `processLoop` by submitting it to `AsyncSupport.getExecutor()`. On JDK <21 the fallback executor is a bounded ThreadPoolExecutor (default max 256) and `processLoop` blocks forever on `queue.take()`, so creating many actors can exhaust the pool and even run the actor loop on the caller thread (CallerRunsPolicy). If the intent is “dedicated thread per actor”, consider starting a new (virtual) thread directly per actor (or using a dedicated executor for actors) rather than the shared async executor.
```suggestion
startActorThread(this::processLoop);
}
private static void startActorThread(Runnable task) {
if (tryStartVirtualThread(task)) {
return;
}
Thread thread = new Thread(task, DefaultActor.class.getSimpleName());
thread.setDaemon(true);
thread.start();
}
private static boolean tryStartVirtualThread(Runnable task) {
try {
java.lang.reflect.Method ofVirtual = Thread.class.getMethod("ofVirtual");
Object builder = ofVirtual.invoke(null);
java.lang.reflect.Method name = builder.getClass().getMethod("name", String.class);
builder = name.invoke(builder, DefaultActor.class.getSimpleName());
java.lang.reflect.Method start = builder.getClass().getMethod("start", Runnable.class);
start.invoke(builder, task);
return true;
} catch (ReflectiveOperationException | RuntimeException ignored) {
return false;
}
}
```
##########
src/main/java/org/apache/groovy/runtime/async/DefaultAsyncScope.java:
##########
@@ -302,6 +320,60 @@ private void pruneCompleted() {
}
}
+ // ---- Child scope management -----------------------------------------
+
+ void registerChildScope(DefaultAsyncScope child) {
+ synchronized (lock) {
+ if (!closed) {
+ childScopes.add(child);
+ }
+ }
+ }
+
+ void deregisterChildScope(DefaultAsyncScope child) {
+ synchronized (lock) {
+ childScopes.remove(child);
+ }
+ }
+
+ // ---- Timeout support ------------------------------------------------
+
+ /**
+ * Creates a scope with a timeout. If the body does not complete within
+ * the duration, all children are cancelled and {@link TimeoutException}
+ * is thrown.
+ */
+ public static <T> T withScopeTimeout(Executor executor, Duration timeout,
+ Function<AsyncScope, T> body) {
+ Objects.requireNonNull(timeout, "timeout must not be null");
+ Objects.requireNonNull(body, "body must not be null");
+ var timedOut = new java.util.concurrent.atomic.AtomicBoolean(false);
+ Thread bodyThread = Thread.currentThread();
+ try (DefaultAsyncScope scope = new DefaultAsyncScope(executor)) {
+ ScheduledFuture<?> timer = AsyncSupport.getScheduler().schedule(() -> {
+ timedOut.set(true);
+ scope.cancelAll();
+ bodyThread.interrupt();
+ }, timeout.toMillis(), TimeUnit.MILLISECONDS);
+ try {
+ T result = withCurrent(scope, () -> body.apply(scope));
+ timer.cancel(false);
+ return result;
+ } catch (Exception e) {
+ timer.cancel(false);
+ if (timedOut.get()) {
+ // Clear the interrupt flag set by the timeout
+ Thread.interrupted();
+ var te = new java.util.concurrent.CancellationException(
+ "Scope timed out after " + timeout);
+ te.initCause(e);
+ throw te;
+ }
+ throw e;
Review Comment:
`withScopeTimeout` only cancels the timer in the success path and when the body throws an `Exception`. If the body throws an `Error` (or other `Throwable` not extending `Exception`), the scheduled timeout task will still fire later, interrupting the thread and cancelling tasks after the scope has already exited. Use a `finally` to always cancel the timer and consider catching `Throwable` (rethrowing) so cleanup is consistent.
```suggestion
return withCurrent(scope, () -> body.apply(scope));
} catch (Throwable t) {
if (timedOut.get()) {
// Clear the interrupt flag set by the timeout
Thread.interrupted();
var te = new java.util.concurrent.CancellationException(
"Scope timed out after " + timeout);
te.initCause(t);
throw te;
}
throw t;
} finally {
timer.cancel(false);
```
##########
src/main/java/org/apache/groovy/runtime/async/DefaultAsyncScope.java:
##########
@@ -302,6 +320,60 @@ private void pruneCompleted() {
}
}
+ // ---- Child scope management -----------------------------------------
+
+ void registerChildScope(DefaultAsyncScope child) {
+ synchronized (lock) {
+ if (!closed) {
+ childScopes.add(child);
+ }
+ }
+ }
+
+ void deregisterChildScope(DefaultAsyncScope child) {
+ synchronized (lock) {
+ childScopes.remove(child);
+ }
+ }
+
+ // ---- Timeout support ------------------------------------------------
+
+ /**
+ * Creates a scope with a timeout. If the body does not complete within
+ * the duration, all children are cancelled and {@link TimeoutException}
+ * is thrown.
+ */
+ public static <T> T withScopeTimeout(Executor executor, Duration timeout,
+ Function<AsyncScope, T> body) {
+ Objects.requireNonNull(timeout, "timeout must not be null");
+ Objects.requireNonNull(body, "body must not be null");
+ var timedOut = new java.util.concurrent.atomic.AtomicBoolean(false);
+ Thread bodyThread = Thread.currentThread();
+ try (DefaultAsyncScope scope = new DefaultAsyncScope(executor)) {
+ ScheduledFuture<?> timer = AsyncSupport.getScheduler().schedule(() -> {
+ timedOut.set(true);
+ scope.cancelAll();
+ bodyThread.interrupt();
+ }, timeout.toMillis(), TimeUnit.MILLISECONDS);
+ try {
+ T result = withCurrent(scope, () -> body.apply(scope));
+ timer.cancel(false);
+ return result;
+ } catch (Exception e) {
+ timer.cancel(false);
+ if (timedOut.get()) {
+ // Clear the interrupt flag set by the timeout
+ Thread.interrupted();
+ var te = new java.util.concurrent.CancellationException(
+ "Scope timed out after " + timeout);
+ te.initCause(e);
+ throw te;
+ }
+ throw e;
Review Comment:
Timeout semantics are inconsistent: the Javadoc says `TimeoutException` is thrown on timeout, but the implementation throws `CancellationException` (and tests catch `CancellationException`). Please align the API contract with the actual behavior (either throw `TimeoutException` as documented, or update the Javadoc in both `AsyncScope.withScope(Duration, ...)` and `DefaultAsyncScope.withScopeTimeout`).
##########
src/main/java/groovy/concurrent/ChannelSelect.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package groovy.concurrent;
+
+import org.apache.groovy.runtime.async.GroovyPromise;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Selects the first available value from multiple {@link AsyncChannel}s.
+ * <p>
+ * This is the channel equivalent of {@link Awaitable#any(Object...)} —
+ * while {@code Awaitable.any} races futures, {@code ChannelSelect} races
+ * channel receives. Each call to {@link #select()} returns an
+ * {@link Awaitable} that completes with a {@link Result} indicating
+ * which channel produced the value and what it was.
+ *
+ * <pre>{@code
+ * def prices = AsyncChannel.create(10)
+ * def alerts = AsyncChannel.create(10)
+ *
+ * def sel = ChannelSelect.from(prices, alerts)
+ * def result = await sel.select()
+ * println "Channel ${result.index}: ${result.value}"
+ * }</pre>
+ * <p>
+ * Inspired by GPars' {@code Select} and Go's {@code select} statement.
+ *
+ * @since 6.0.0
+ */
+public final class ChannelSelect {
+
+ private final List<AsyncChannel<?>> channels;
+
+ private ChannelSelect(List<AsyncChannel<?>> channels) {
+ this.channels = channels;
+ }
+
+ /**
+ * Creates a select over the given channels.
+ *
+ * @param channels the channels to select from
+ * @return a new ChannelSelect
+ */
+ @SafeVarargs
+ public static ChannelSelect from(AsyncChannel<?>... channels) {
+ Objects.requireNonNull(channels, "channels must not be null");
+ if (channels.length == 0) {
+ throw new IllegalArgumentException("At least one channel is required");
+ }
+ return new ChannelSelect(List.of(channels));
+ }
+
+ /**
+ * Waits for the first value available from any of the channels.
+ * <p>
+ * Returns an {@link Awaitable} that completes with a {@link Result}
+ * containing the channel index and the received value.
+ * <p>
+ * Values consumed by non-winning channels are re-sent back to those
+ * channels to prevent message loss. This may reorder values within
+ * a channel but guarantees no values are silently dropped.
+ *
+ * @return an awaitable result indicating which channel produced the value
+ */
+ @SuppressWarnings("unchecked")
+ public Awaitable<Result> select() {
+ CompletableFuture<Result> winner = new CompletableFuture<>();
+ AtomicBoolean won = new AtomicBoolean();
+ for (int i = 0; i < channels.size(); i++) {
+ final int index = i;
+ AsyncChannel<?> ch = channels.get(i);
+ ch.receive().toCompletableFuture().whenComplete((value, error) -> {
+ if (error != null) return;
+ if (won.compareAndSet(false, true)) {
+ winner.complete(new Result(index, value));
+ } else {
+ // Re-send the consumed value back to avoid message loss
+ try {
+ ((AsyncChannel<Object>) ch).send(value);
+ } catch (ChannelClosedException ignored) {
+ // Channel was closed; value cannot be preserved
+ }
+ }
+ });
Review Comment:
`select()` ignores errors from `receive()` (including `ChannelClosedException`). If one or more channels are closed/drained, their callbacks just `return`, and if all channels close without producing a value the `winner` future never completes, causing an infinite await/hang. Track closed/error completions and complete `winner` exceptionally once all channels have terminated (and consider propagating non-closed errors immediately).
--
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: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org