[
https://issues.apache.org/jira/browse/GROOVY-12320?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Paul King updated GROOVY-12320:
-------------------------------
Description:
{{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.
was:
{{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 a fair (rotating) policy is 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: {{ChannelSelect.fair()}} — since a ready channel completes
synchronously during registration, rotating the registration start to the
channel after the last winner gives JCSP-style fair selection; the default
stays priority by list order and is documented as such.
> 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)
