[
https://issues.apache.org/jira/browse/GROOVY-12320?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18109555#comment-18109555
]
ASF GitHub Bot commented on GROOVY-12320:
-----------------------------------------
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
> ChannelSelect.select() consumes from losing branches; add fair selection
> policy
> -------------------------------------------------------------------------------
>
> Key: GROOVY-12320
> URL: https://issues.apache.org/jira/browse/GROOVY-12320
> Project: Groovy
> Issue Type: Bug
> Reporter: Paul King
> Assignee: Paul King
> Priority: Major
>
> {{ChannelSelect.select()}} is implemented by racing a real {{receive()}} on
> every channel and completing with the first to deliver; the losing branches
> re-{{send}} what they took. Because losers genuinely dequeue, a select is not
> a choice between channels but a consumption from all of them, and several
> things follow. All reproduce on 6.0.0-beta-3 (unchanged since GEP-18, so
> alpha-1 onwards).
> h3. Observed
> # *A losing channel's order is broken.* The value a loser took is put back
> with {{send}}, i.e. at the _back_ of the queue. With {{b}} holding {{[b1,
> b2]}}, one select over {{[a, b]}} that {{a}} wins leaves {{b}} delivering
> {{[b2, b1]}}. A consumer of {{b}} cannot know someone selected over {{b}}, so
> it cannot reason about its own input order.
> # *Every select leaves a pending receiver on each branch that was empty at
> the time.* Nothing cancels them. 1000 selects over {{[busy, quiet]}} with
> {{quiet}} always empty leave 1000 entries in {{quiet}}'s {{waitingReceivers}}.
> # *Those stale receivers lose the next message.* When {{quiet}} finally gets
> a value, each stale receiver takes it and re-sends it, synchronously, under
> the channel lock — a recursion as deep as the number of stale receivers. Past
> a stack-dependent threshold the {{StackOverflowError}} is caught by
> {{CompletableFuture}} as an action failure; the sender's {{send}} returns
> success and the value is gone. Measured with 5000 stale receivers:
> {{-Xss256k}} → 74 bounces then lost, {{1m}} → 494, {{2m}} (default) → ~1024,
> {{32m}} → settles. So a select loop that has run ~1000 iterations since its
> quiet branch last fired silently drops that branch's next message — and quiet
> branches are where shutdown/control messages live.
> # *A timed-out select drops the next value.* {{select().orTimeoutMillis\(n)}}
> cancels the winner future but the branch receives stay registered; the first
> value to arrive wins the internal CAS, is handed to the cancelled future
> ({{complete}} returns false) and is discarded.
> # *A stale receiver breaks rendezvous.* On an unbuffered channel a losing
> select receiver accepts the value, so {{send}} completes with no consumer;
> the re-send becomes a pending sender nobody holds, and {{close()}} loses it.
> # *A select over channels that are all closed and drained never completes.*
> Closed branches complete exceptionally and the callback ignores errors;
> nothing completes the winner.
> The Javadoc claim "guarantees no values are silently dropped" is therefore
> not true; the {{catch (ChannelClosedException)}} around the re-send is dead
> code ({{send}} never throws, it returns a failed Awaitable that is discarded).
> Separately, when several channels are ready the first-listed one always wins
> (a ready branch's {{receive()}} completes synchronously inside the
> registration loop). That is a priority select, which is fine but
> undocumented, and there is no fair alternative: a channel that always has a
> value waiting starves the channels after it, the case JCSP's {{fairSelect}}
> and Go's random choice exist to prevent.
> h3. Steps to reproduce
> {code:groovy}
> import groovy.concurrent.*
> import static org.apache.groovy.runtime.async.AsyncSupport.*
> // 1. reordering
> def a = AsyncChannel.<String>create(4), b = AsyncChannel.<String>create(4)
> a.send('a1'); b.send('b1'); b.send('b2')
> await ChannelSelect.from(a, b).select() // a wins
> assert [await(b.receive()), await(b.receive())] == ['b1', 'b2'] // FAILS:
> [b2, b1]
> // 2/3. stale receivers, then loss
> def busy = AsyncChannel.<Integer>create(4), quiet =
> AsyncChannel.<Integer>create(4)
> 5000.times { busy.send(it); await ChannelSelect.from(busy, quiet).select() }
> quiet.send(7); Thread.sleep(300)
> assert quiet.bufferedSize == 1 // FAILS: 0 — the
> value is gone
> // 4. timeout
> def c = AsyncChannel.<String>create(4), d = AsyncChannel.<String>create(4)
> try { await ChannelSelect.from(c, d).select().orTimeoutMillis(100) } catch
> (java.util.concurrent.TimeoutException ignore) {}
> c.send('x'); c.send('y')
> assert (await c.receive()) == 'x' // FAILS: 'y' — 'x'
> was consumed by the dead select
> // 5. rendezvous
> def r = AsyncChannel.<Integer>create()
> busy.send(1); await ChannelSelect.from(busy, r).select()
> assert !r.send(42).toCompletableFuture().isDone() // FAILS: completed
> with no receiver
> // 6. all closed
> def e = AsyncChannel.<Integer>create(1), f = AsyncChannel.<Integer>create(1)
> e.close(); f.close()
> await ChannelSelect.from(e, f).select().orTimeoutMillis(500) //
> TimeoutException, not ChannelClosedException
> {code}
> h3. Expected
> * A select takes exactly one value from exactly one channel; the other
> channels are untouched (contents and order preserved, nothing left registered
> on them once the result completes).
> * A cancelled or timed-out select consumes nothing.
> * A losing branch on an unbuffered channel never completes a send.
> * A select over channels that are all closed and drained fails with
> {{ChannelClosedException}}.
> * The tie-break policy is documented, and fair (rotating) and random policies
> are available.
> h3. Proposed fix
> Select by _claim_, not by consumption (the enable/wait/disable discipline of
> JCSP's {{Alternative}}):
> * {{DefaultAsyncChannel}} gains an internal
> {{receiveIfUnclaimed(AtomicBoolean claim)}}. At every hand-over point the
> channel, under its lock, does _claim → complete → dequeue on success_; a
> receiver that loses the claim is discarded and the value stays in place.
> * {{ChannelSelect.select()}} shares one claim across its branches, withdraws
> (cancels) the losing branches as soon as one wins or the result is cancelled,
> and fails with {{ChannelClosedException}} once every branch has failed. The
> waiting-receiver queue becomes a concurrent deque so that withdrawal never
> takes a second channel's lock from inside a delivery (two concurrent selects
> over {{(A, B)}} / {{(B, A)}} could otherwise deadlock).
> * {{orTimeout}} / {{completeOnTimeout}} cancel the source before completing
> the result, so a consuming source cannot take a value the caller will never
> see.
> * Follow-up commit: since a ready channel completes synchronously during
> registration, the registration order is the tie-break order, so alternative
> policies only change where the loop starts. {{ChannelSelect.fair()}} starts
> at the channel after the last winner (JCSP {{fairSelect}}: every ready
> channel is taken within n calls); {{random()}} starts at a uniformly random
> channel (Go's {{select}}, GPars' {{Select}}: stateless, equally fair from any
> number of threads, no bound on waiting). The default stays priority by list
> order and is documented as such.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
