public final class ChannelSelect
extends ObjectSelects the first offer that can complete among multiple AsyncChannel operations.
This is the channel equivalent of Awaitable.any —
while Awaitable.any races futures, ChannelSelect races
channel operations. Each call to select() returns an
Awaitable that completes with a Result indicating
which offer committed and, for a receive, what value arrived.
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}"
A select may also mix input and output guards — the mixed choice of the CSP literature: "I will send my opener, or take my peer's if it sends first". Offers to send are created with send(AsyncChannel, Object), offers to receive with receive(AsyncChannel), and combined with offers(Offer...):
import static groovy.concurrent.ChannelSelect.*
def result = await offers(send(ping, 1), receive(pong)).select()
if (result.send) { ... my opener committed ... }
else { ... my peer opened first: result.value ... }
A send offer commits as soon as its channel can accept the value: when a receiver is waiting, or when buffer space holds it. Two peers racing a mixed choice are therefore coherent — exactly one of the two openers commits — only over rendezvous (capacity-0) channels, where a send cannot complete unilaterally; over buffered channels with space, each peer's send offer commits immediately under its own select, and the two proceed down different branches of the same session.
Any branch may also carry a precondition — the guarded choice of CSP — either written onto the offer itself with Offer.when or passed positionally to select(boolean...). A disabled offer is left unregistered while the enabled ones keep their positions, so a guard can be masked off without renumbering the branches around it.
def result = await offers(receive(input).when { size < capacity },
receive(request).when { size > 0 }).select()
A deadline is just another branch. The timer offer after(long) arms a fresh timer on every select() call, so a select held across rounds waits for work, but not forever, and keeps its choice policy state while it does:
def sel = offers(receive(work), after(100)).fair()
while (true) {
def result = await sel.select()
if (result.timeout) { ... nothing arrived this round ... }
}
Inspired by GPars' Select and Go's select statement
(whose send cases these offers mirror).
| Modifiers | Name | Description |
|---|---|---|
static class |
ChannelSelect.Offer |
One branch of a select: an input guard created by receive(AsyncChannel), an output guard created by send(AsyncChannel, Object), or a timeout created by after(long). |
static class |
ChannelSelect.Result |
The result of a select() operation, indicating which offer committed. |
| Type Params | Return Type | Name and description |
|---|---|---|
|
public static ChannelSelect.Offer |
after(long millis)An offer that commits once millis has elapsed from the start
of the select() call it takes part in: a timeout as a branch
of the choice, rather than an exception thrown around it by
Awaitable.orTimeout. |
|
public static ChannelSelect.Offer |
after(Duration duration)An offer that commits once duration has elapsed from the
start of the select call it takes part in; see after(long). |
|
public ChannelSelect |
fair()Returns a select over the same offers that chooses fairly among offers that are ready at the same time. |
|
public static ChannelSelect |
from(AsyncChannel<?> channels)Creates a select over receives from the given channels. |
|
public static ChannelSelect |
offers(ChannelSelect.Offer offers)Creates a select over the given offers, which may mix sends and receives. |
|
public ChannelSelect |
random()Returns a select over the same offers that chooses uniformly at random among offers that are ready at the same time. |
|
public static ChannelSelect.Offer |
receive(AsyncChannel<?> channel)An offer to receive the next value from channel: an input
guard, the branch a plain from(AsyncChannel...) select is
made of. |
|
public Awaitable<Result> |
select()Waits for the first offer that can complete. |
|
public Awaitable<Result> |
select(boolean enabled)Waits for the first offer that can complete among those its precondition enables: the guarded choice of the CSP literature, where a branch is masked off for this call without disturbing the others. |
<V> |
public static ChannelSelect.Offer |
send(AsyncChannel<V> channel, V value)An offer to transfer value into channel: an output
guard. |
An offer that commits once millis has elapsed from the start
of the select() call it takes part in: a timeout as a branch
of the choice, rather than an exception thrown around it by
Awaitable.orTimeout. It commits with the
Instant at which it fired.
Semantically it is a receive from a timer channel that the select
creates and, if the offer loses, cancels on every call — the
difference from receive(AsyncChannel.after(millis)), whose
clock starts once, when the channel is created. The timer offer is
therefore the "wait for work, but not forever" branch of a select
that is held and reused, where it re-arms each round while the
instance keeps its fair() rotation. The channel form is
the fixed deadline shared by every round. Like any other offer it
may be guarded with Offer.when; a guarded-off timer offer
arms nothing.
A timer offer with a positive delay is never ready at the moment of
registration, so among offers that are ready at once the choice
policy considers only the channel offers; the timer commits only when
no channel offer could before it fired. A delay that is not positive
has already elapsed, so that offer is ready at registration and takes
part in the choice among ready offers like any other: under the
default priority it wins if listed before every ready channel offer
and loses to one listed before it, and fair() rotates over it.
Listed last, after(0) is therefore the default clause
of Go's select: take a transfer if one can complete at once,
otherwise proceed without waiting.
def result = await offers(receive(work), after(0)).select()
if (result.timeout) { ... nothing was ready ... }
millis - how long to wait, in milliseconds, from the start of
each select call An offer that commits once duration has elapsed from the
start of the select call it takes part in; see after(long).
duration - how long to wait from the start of each select callReturns a select over the same offers that chooses fairly among offers that are ready at the same time.
By default select() prefers the offer listed first, so a
channel that always has a value waiting starves the ones after it. A
fair select instead starts each call at the offer after the one
that last won, so every offer that is ready is taken within
n calls, where n is the number of offers. This is
the rotating policy of JCSP's fairSelect. When no offer is
ready, the first to become completable wins under either policy.
The rotation state lives in the returned instance, so keep and reuse it across calls (typically in a loop); a shared instance may be used from several threads, in which case the rotation is best effort.
Creates a select over receives from the given channels.
channels - the channels to select fromCreates a select over the given offers, which may mix sends and receives.
offers - the offers to select amongReturns a select over the same offers that chooses uniformly at random among offers that are ready at the same time.
This is the policy of Go's select and GPars' Select.
Unlike fair() it keeps no state between calls, so it is
exactly as fair from any number of threads and cannot fall into
lock-step with the producers, but it offers no bound on how long a
ready offer may be passed over. When no offer is ready, the first
to become completable wins under either policy.
An offer to receive the next value from channel: an input
guard, the branch a plain from(AsyncChannel...) select is
made of.
channel - the channel to receive fromWaits for the first offer that can complete.
Returns an Awaitable that completes with a Result naming the committed offer: the received value for an input guard, the sent value for an output guard.
Exactly one offer commits. The other channels are left untouched: their contents and order are preserved, and nothing remains registered on them once the result completes — a retired send offer in particular leaves no buffered residue and no waiting sender. When several offers are ready, the one listed first commits (see fair() for a rotating choice and random() for a random one). Cancelling the result (for example through Awaitable.orTimeout) withdraws the pending offers, so a timed-out select consumes nothing and sends nothing. To take a timeout as a branch instead of an exception, add a timer offer from after(long) or receive from a timer channel from AsyncChannel.after.
If every offer's channel is closed, the result fails with ChannelClosedException. If every offer is guarded off (see Offer.when) there is nothing to wait for and the result fails with IllegalStateException.
Only channels created by AsyncChannel.create take part in the
claim protocol that makes this possible. Any other AsyncChannel
implementation consumes a value before the select can decide; if that
value loses, it is re-sent to its channel, which preserves it but may
reorder that channel. Send offers are limited to built-in channels for
the same reason.
Waits for the first offer that can complete among those its precondition enables: the guarded choice of the CSP literature, where a branch is masked off for this call without disturbing the others.
Flag i enables offer i, and is conjoined with any guard
the offer carries of its own: an offer is registered only if both hold.
A disabled offer is not registered on its channel — nothing is consumed
from it and nothing is sent to it — but it keeps its position, so
Result.getIndex still denotes the same branch whatever the
mask. That positional stability is the point: dropping an offer from
the argument list instead would silently renumber the branches after it.
// the classic bounded buffer: take input only while there is room,
// answer requests only while there is something to hand over
def sel = offers(receive(input), receive(request))
while (true) {
def result = await sel.select(size < capacity, size > 0)
if (result.index == 0) { ... buffer result.value ... }
else { ... reply with the oldest value ... }
}
In every other respect this behaves as select(): the enabled offers are scanned in the order the choice policy dictates, and exactly one commits.
enabled - one flag per offer, in offer order An offer to transfer value into channel: an output
guard. It commits when the channel can accept the value — a waiting
receiver takes it, or buffer space holds it — and a committed offer
behaves exactly like channel.send(value). A retired offer has
no effect on the channel.
Only channels created by AsyncChannel.create can take part in the claim protocol that makes an effect-free retired send possible, so only they may carry send offers.
channel - the channel to send intovalue - the value to transferV - the payload type