Package groovy.concurrent

Class ChannelSelect

java.lang.Object
groovy.concurrent.ChannelSelect
public final class ChannelSelect extends Object
Selects the first offer that can complete among multiple AsyncChannel operations.

This is the channel equivalent of Awaitable.any(Object...) — while Awaitable.any races futures, ChannelSelect races channel operations. Each call to select() returns an Awaitable that completes with a ChannelSelect.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 ChannelSelect.Offer.when(BooleanSupplier) 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 ... }
 }
For a fixed deadline across rounds, receive from a timer channel instead: AsyncChannel.after(long) starts its clock when it is created and delivers the instant it fired.

Inspired by GPars' Select and Go's select statement (whose send cases these offers mirror).

Since:
6.0.0

  • Method Details

    • from

      @SafeVarargs public static ChannelSelect from(AsyncChannel<?>... channels)
      Creates a select over receives from the given channels.
      Parameters:
      channels - the channels to select from
      Returns:
      a new ChannelSelect

    • offers

      public static ChannelSelect offers(ChannelSelect.Offer... offers)
      Creates a select over the given offers, which may mix sends and receives.
      Parameters:
      offers - the offers to select among
      Returns:
      a new ChannelSelect
      Since:
      6.0.0
      See Also:

    • send

      public static <V> ChannelSelect.Offer send(AsyncChannel<V> channel, V value)
      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.

      Type Parameters:
      V - the payload type
      Parameters:
      channel - the channel to send into
      value - the value to transfer
      Returns:
      the send offer
      Throws:
      IllegalArgumentException - if the channel is not a built-in one
      Since:
      6.0.0

    • receive

      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.
      Parameters:
      channel - the channel to receive from
      Returns:
      the receive offer
      Since:
      6.0.0

    • after

      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(long, TimeUnit). 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 ChannelSelect.Offer.when(java.util.function.BooleanSupplier); 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 ... }
      Parameters:
      millis - how long to wait, in milliseconds, from the start of each select call
      Returns:
      the timer offer
      Since:
      6.0.0
      See Also:

    • after

      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).
      Parameters:
      duration - how long to wait from the start of each select call
      Returns:
      the timer offer
      Throws:
      NullPointerException - if duration is null
      Since:
      6.0.0

    • fair

      public ChannelSelect fair()
      Returns 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.

      Returns:
      a fair ChannelSelect over the same offers
      Since:
      6.0.0
      See Also:

    • random

      public ChannelSelect random()
      Returns 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.

      Returns:
      a randomly choosing ChannelSelect over the same offers
      Since:
      6.0.0
      See Also:

    • select

      public Awaitable<ChannelSelect.Result> select()
      Waits for the first offer that can complete.

      Returns an Awaitable that completes with a ChannelSelect.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(long, java.util.concurrent.TimeUnit)) 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(long).

      If every offer's channel is closed, the result fails with ChannelClosedException. If every offer is guarded off (see ChannelSelect.Offer.when(BooleanSupplier)) 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.

      Returns:
      an awaitable result indicating which offer committed

    • select

      public Awaitable<ChannelSelect.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.

      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 ChannelSelect.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.

      Parameters:
      enabled - one flag per offer, in offer order
      Returns:
      an awaitable result indicating which offer committed, or one that fails with IllegalStateException if every offer is disabled
      Throws:
      IllegalArgumentException - if the number of flags is not the number of offers
      Since:
      6.0.0
      See Also: