Copilot commented on code in PR #2846:
URL: https://github.com/apache/groovy/pull/2846#discussion_r3888017384
##########
src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java:
##########
@@ -448,14 +449,19 @@ public static Awaitable<Void> delay(long duration,
TimeUnit unit) {
public static <T> Awaitable<T> orTimeout(Object source, long timeout,
TimeUnit unit) {
CompletableFuture<T> future = (CompletableFuture<T>)
Awaitable.from(source).toCompletableFuture();
CompletableFuture<T> result = new CompletableFuture<>();
+ AtomicBoolean timedOut = new AtomicBoolean();
ScheduledFuture<?> timer = AsyncExecutors.getScheduler().schedule(()
-> {
- if (!result.isDone()) {
- result.completeExceptionally(new TimeoutException("Timed out
after " + timeout + " " + unit));
+ if (!result.isDone() && timedOut.compareAndSet(false, true)) {
+ // withdraw the source before failing the result, so a
consuming
+ // source (a channel receive or select) cannot take a value the
+ // caller will never see
future.cancel(true);
+ result.completeExceptionally(new TimeoutException("Timed out
after " + timeout + " " + unit));
Review Comment:
`CompletableFuture.cancel` may synchronously run arbitrary dependent
callbacks, so placing it before completing `result` can prevent this timeout
from ever firing if one such callback blocks (or waits on `result`). Complete
the timeout independently of source-cancellation callbacks; preserving the
channel withdrawal ordering requires coordinated cancellation in the
channel/select implementation rather than blocking the scheduler here.
This issue also appears on line 487 of the same file.
##########
src/main/java/groovy/concurrent/ChannelSelect.java:
##########
@@ -76,36 +130,81 @@ public static ChannelSelect from(AsyncChannel<?>...
channels) {
* 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.
+ * Exactly one value is taken, from exactly one channel. The other
+ * channels are left untouched: their contents and order are preserved,
+ * and nothing remains registered on them once the result completes.
+ * When several channels already hold a value, the one listed first is
+ * taken (see {@link #fair()} for a rotating choice and {@link #random()}
+ * for a random one). Cancelling the
+ * result (for example through
+ * {@link Awaitable#orTimeout(long, java.util.concurrent.TimeUnit)})
+ * withdraws the pending receives, so a timed-out select consumes
+ * nothing.
+ * <p>
+ * If every channel is closed and drained, the result fails with
+ * {@link ChannelClosedException}.
+ * <p>
+ * Only channels created by {@link AsyncChannel#create} take part in the
+ * claim protocol that makes this possible. For other {@code AsyncChannel}
+ * implementations a value consumed by a losing branch is re-sent to its
+ * channel, which preserves it but may reorder that channel.
*
* @return an awaitable result indicating which channel produced the value
*/
- @SuppressWarnings("unchecked")
public Awaitable<Result> select() {
+ int count = channels.size();
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
+ AtomicBoolean claim = new AtomicBoolean();
+ AtomicInteger closedCount = new AtomicInteger();
+ Awaitable<?>[] branches = new Awaitable<?>[count];
+
+ // a ready channel completes synchronously during registration, so the
+ // registration order is the priority order: rotate it under fair(),
+ // start it anywhere under random()
+ int start = switch (policy) {
+ case PRIORITY -> 0;
+ case FAIR -> Math.floorMod(lastWinner.get() + 1, count);
+ case RANDOM -> ThreadLocalRandom.current().nextInt(count);
+ };
+ for (int k = 0; k < count && !winner.isDone(); k++) {
+ final int index = (start + k) % count;
Review Comment:
`random()` does not choose uniformly among the ready channels. Choosing one
random start and scanning cyclically weights each ready channel by the size of
the non-ready gap before it; for example, with channels 0 and 1 ready out of
three, channel 0 wins for starts 0 and 2 (2/3) while channel 1 wins only for
start 1 (1/3). Randomize the full registration order instead so every ready
channel has the same chance to be encountered first.
This issue also appears in the following locations of the same file:
- line 177
- line 179
--
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]
