Generators, Channels & Structured Concurrency

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production.

The reference material consulted while preparing these pages is listed in the section bibliography.

This page is written against Apache Groovy 6.0 (6.0.0, 2026-09-22) and every example was run with Groovy 6.0.0 on JDK 26 (checked on 2026-10-14). Everything here is part of the concurrency toolkit that is new and still incubating in Groovy 6.0, so the APIs in groovy.concurrent may change in a minor release. It builds on the async and await keywords introduced on Async/Await and adds five things: generators that produce a sequence of values lazily (yield return), for await to consume any asynchronous source, Go-style channels for communication between tasks, defer for guaranteed cleanup, and AsyncScope so that the lifetime of child tasks is bound to a block. On JDK 21 and later the tasks run on virtual threads, which is what makes blocking inside a producer or consumer cheap; on JDK 17 to 20 a cached thread pool is used instead. As on the previous page, the types need import groovy.concurrent.* even though the official guide leaves the import out. The running scenario is the greenhouse: sensor readings flowing from producers to consumers. The same ideas exist in Kotlin as Kotlin Flows and Coroutines, and Java 21 provides the underlying threads on Java Virtual Threads.

Generators with yield return

An async closure that contains yield return is not an ordinary async task but a generator: it produces its values one by one, on demand, and what the expression returns is a plain java.lang.Iterable, not an Awaitable. The generator body runs on a background thread and pauses at each yield return until the consumer asks for the next element, which gives natural back-pressure and lets the sequence be infinite. Regular for loops and the GDK collection methods (collect, findAll, take…) work on it unchanged.

import groovy.concurrent.*

// A generator: an async closure containing "yield return"
def dayReadings = {
    async {
        for (hour in 0..<24) {
            yield return [hour: hour, celsius: 18 + (hour % 12)]
        }
    }
}
def readings = dayReadings()
println readings.getClass().interfaces*.simpleName
println readings.findAll { it.celsius >= 28 }
println "second pass over the same generator: ${readings.collect { it.celsius }}"
println dayReadings().collect { it.celsius }.max()    // a fresh generator per pass

// Lazy and infinite: only what is consumed is produced
def fib = async {
    long a = 0, b = 1
    while (true) {
        yield return a
        (a, b) = [b, a + b]
    }
}
println fib.take(8).collect()
println fib.take(3).collect()        // continues where the first take stopped

// Back-pressure: the producer pauses at each yield return until the consumer asks
def trace = Collections.synchronizedList([])
def gen = async {
    for (i in 1..3) { trace << "produce $i"; yield return i }
    trace << 'producer done'
}
for (v in gen) { trace << "consume $v" }
println trace

[Iterable]
[[hour:10, celsius:28], [hour:11, celsius:29], [hour:22, celsius:28], [hour:23, celsius:29]]
second pass over the same generator: []
29
[0, 1, 1, 2, 3, 5, 8, 13]
[21, 34, 55]
[produce 1, produce 2, consume 1, consume 2, produce 3, consume 3, producer done]

Source: Groovy documentation: Async/await, Generators and streaming

Three behaviours matter in practice:

A generator is a one-shot sequence. The second pass over readings printed an empty list because the first pass had already consumed it, and fib.take(3) continued where fib.take(8) stopped (21, 34, 55). To iterate again, create a new generator, which is why a factory closure such as dayReadings is the usual shape.
The producer is paused by the consumer. In the trace, produce 3 appears only after consume 2: the generator is never more than a step ahead, so it cannot overrun a slow consumer or build an unbounded buffer.
An exception thrown inside the generator body is delivered to the consumer at the point where the failing value would have been read, as the next example shows.

import groovy.concurrent.*
// an exception in a generator surfaces in the consumer
def flaky = async {
    yield return 1
    yield return 2
    throw new IllegalStateException('sensor bus lost')
}
def got = []
try { for await (v in flaky) { got << v } } catch (IllegalStateException e) { println "after $got: ${e.message}" }

after [1, 2]: sensor bus lost

Source: Groovy documentation: Async/await, Producing values with yield return

A generator is the right choice when producing the next value involves blocking, such as paging an API or tailing a log. For pure in-memory laziness, plain closures and collection methods (see Functional Programming) need no threads at all.

Consuming with for await

for await (x in source) iterates over any asynchronous source. For a generator, a channel or a plain collection it behaves like an ordinary for loop. Its real value is with reactive types: given a Flow.Publisher (supported out of the box), a Reactor Flux (module groovy-reactor) or an RxJava Observable or Flowable (module groovy-rxjava), it converts the source to a blocking iterable through the adapter SPI for you, so you do not have to call flux.toIterable() yourself. The adapters are introduced on Async/Await.

import groovy.concurrent.*
def sensorIds = {
    async { for (s in ['gh-1', 'gh-2', 'gh-3']) { yield return s } }
}
def seen = []
for await (s in sensorIds()) { seen << s }
println seen
// for await also works on plain collections and ranges
def squares = []
for await (n in [1, 2, 3]) { squares << n * n }
println squares
// a generator can be abandoned early with break
for await (s in sensorIds()) { if (s == 'gh-2') break; println "visited $s" }

[gh-1, gh-2, gh-3]
[1, 4, 9]
visited gh-1

Source: Groovy documentation: Async/await, Consuming with for await

The last loop leaves a generator unfinished with break; because a generator is one-shot, only the elements before the break were visited. The Agent.changes() stream described on Actors, Agents & Active Objects is another publisher that for await consumes directly.

Channels

A channel, AsyncChannel, is a Go-style conduit between tasks. One task sends values, another receives them, and the channel takes care of synchronisation and optional buffering, so no shared mutable state is needed. AsyncChannel.create() makes an unbuffered (rendezvous) channel on which a send completes only when a receiver takes the value; AsyncChannel.create(n) makes a buffered one with n slots, where send suspends only when the buffer is full and receive only when it is empty. send and receive both return an Awaitable, so you await them, and close() signals that no more values will come. A channel also implements Iterable, so for await, for and the GDK methods consume it until it is closed and drained.

import groovy.concurrent.*

// Buffered channel: producer and consumer decoupled by 3 slots
def readings = AsyncChannel.create(3)
println "capacity=${readings.capacity}"
async {
    for (r in [20.5, 21.0, 21.5, 22.0]) { await readings.send(r) }
    readings.close()
}
def got = []
for await (r in readings) { got << r }
println got
println "closed=${readings.closed}"

// receive() returns an Awaitable; after close and drain it fails
def ch = AsyncChannel.create()                // unbuffered (rendezvous)
async { await ch.send('ping'); ch.close() }
println await(ch.receive())
try { await ch.receive() } catch (ChannelClosedException e) { println "closed: ${e.class.simpleName}" }

// Channels are Iterable, so plain loops and GDK methods work
def c2 = AsyncChannel.create(5)
async { (1..5).each { await c2.send(it) }; c2.close() }
println c2.collect { it * 10 }

capacity=3
[20.5, 21.0, 21.5, 22.0]
closed=true
ping
closed: ChannelClosedException
[10, 20, 30, 40, 50]

Source: Groovy documentation: Async/await, Channels

Closing has precise semantics, summarised by the next run. close() returns true the first time and false afterwards; sending to a closed channel fails with ChannelClosedException; values already buffered stay readable after the close, and a receive on a closed, drained channel also fails with ChannelClosedException, which is how consumers know to stop when they do not use for await.

import groovy.concurrent.*

// sending to a closed channel fails
def ch = AsyncChannel.create(2)
println ch.close()
println ch.close()
try { await ch.send(1) } catch (ChannelClosedException e) { println "send after close: ${e.class.simpleName}: ${e.message}" }
// buffered values survive close
def c2 = AsyncChannel.create(3)
await c2.send('a'); await c2.send('b')
println "buffered=${c2.bufferedSize}"
c2.close()
println c2.collect()

true
false
send after close: ChannelClosedException: channel is closed for send
buffered=2
[a, b]

Source: Groovy documentation: Async/await, Channels

Channel composition

Composition methods build pipelines. Each one returns a new channel connected to its source by a background task, so nothing is copied eagerly and the stages run concurrently:

Method Result

Method	Result
`filter(predicate)`	a channel with only the values the predicate accepts
`map(function)`	a channel with each value transformed
`merge(other)`	one channel that interleaves both sources as values arrive
`split(predicate)`	a list of two channels: values matching the predicate, and the rest
`tap(monitor)`	a channel that passes everything through and sends a copy of each value to `monitor`

filter(predicate)

a channel with only the values the predicate accepts

map(function)

a channel with each value transformed

merge(other)

one channel that interleaves both sources as values arrive

split(predicate)

a list of two channels: values matching the predicate, and the rest

tap(monitor)

a channel that passes everything through and sends a copy of each value to monitor

Operations chain naturally, for example source.filter { it > 50 }.map { it * 2 }.

flowchart LR S["sensor feed"] --> F["filter: above 30"] F --> M["map: format alert"] M --> T["tap"] T --> O["alert consumer"] T -. copy .-> L["monitor channel"]

import groovy.concurrent.*

// helper: a channel fed from a list by a background task
def feed = { List items, int cap = 10 ->
    def ch = AsyncChannel.create(cap)
    async { items.each { await ch.send(it) }; ch.close() }
    ch
}
def drain = { ch -> def out = []; for await (v in ch) { out << v }; out }

// filter and map; chaining
def temps = feed([17.5, 22.0, 31.5, 25.0, 35.0, 19.0])
def alerts = temps.filter { it > 30 }.map { "ALERT ${it}C" }
println drain(alerts)

// merge: interleave two sources as values arrive
def east = feed([1, 2, 3]); def west = feed([10, 20, 30])
println drain(east.merge(west)).sort()

// split: partition by predicate into two channels
def (hot, ok) = feed([18.0, 33.0, 21.0, 36.5, 24.0]).split { it > 30 }
def hotList = async { drain(hot) }
def okList = async { drain(ok) }
println "hot=${await hotList} ok=${await okList}"

// tap: pass values through and send a copy to a monitor channel
def monitor = AsyncChannel.create(100)
def out = feed(['a', 'b', 'c']).tap(monitor)
println "output=${drain(out)}"
monitor.close()
println "monitor=${drain(monitor)}"

[ALERT 31.5C, ALERT 35.0C]
[1, 2, 3, 10, 20, 30]
hot=[33.0, 36.5] ok=[18.0, 21.0, 24.0]
output=[a, b, c]
monitor=[a, b, c]

Source: Groovy documentation: Async/await, Channel composition

Two details of the example: merge makes no ordering promise between its two sources, which is why the result is sorted before printing, and both halves of a split are consumed here by two separate async tasks. Give every channel in a pipeline a consumer: the stages are connected by bounded channels, and a result nobody reads leaves its producer waiting.

ChannelSelect

ChannelSelect waits on several channels and takes exactly one value from exactly one of them, like select in Go. The losing channels are left untouched. select() returns an Awaitable of a result with the index of the winning branch, its value, the channel and the flags send and timeout. When several channels are ready at once, the one listed first wins, which can starve the later ones; fair() returns a select that rotates its starting point after every win (keep the instance across calls), and random() picks uniformly among the ready channels. When every channel is closed and drained, select() fails with ChannelClosedException.

import groovy.concurrent.*

def prices = AsyncChannel.create(10)
def alerts = AsyncChannel.create(10)
alerts.send('frost warning')
prices.send(4.2)
def sel = ChannelSelect.from(prices, alerts)
def r1 = await sel.select()
println "first:  index=${r1.index} value=${r1.value}"      // listed first wins when both are ready
def r2 = await sel.select()
println "second: index=${r2.index} value=${r2.value}"

// fair(): rotate the starting channel so a busy channel cannot starve the others
def busy = AsyncChannel.create(10)
def control = AsyncChannel.create(10)
5.times { busy.send("work-$it") }
control.send('stop')
def fairSel = ChannelSelect.from(busy, control).fair()
def order = []
3.times { order << (await fairSel.select()).value }
println "fair order: $order"

// priority select would starve control
def busy2 = AsyncChannel.create(10)
def control2 = AsyncChannel.create(10)
5.times { busy2.send("work-$it") }
control2.send('stop')
def prio = ChannelSelect.from(busy2, control2)
def order2 = []
3.times { order2 << (await prio.select()).value }
println "priority order: $order2"

// closed and drained: select fails
def a = AsyncChannel.create(); a.close()
try { await ChannelSelect.from(a).select() } catch (ChannelClosedException e) { println "all closed: ${e.class.simpleName}" }

first:  index=0 value=4.2
second: index=1 value=frost warning
fair order: [work-0, stop, work-1]
priority order: [work-0, work-1, work-2]
all closed: ChannelClosedException

Source: Groovy documentation: Async/await, Channel select

The fair select served control after one busy value, while the priority select kept serving busy and would have left control waiting for as long as busy had work.

Mixed offers

ChannelSelect.offers(...) accepts offers built with the static factories send(channel, value) and receive(channel), the mixed choice of CSP: "send my opener, or take my peer’s if it sends first". Exactly one offer commits. A committed send behaves like channel.send(value); a losing offer leaves no trace. Over rendezvous channels, a send offer commits only when a receiver is waiting, so two peers racing a mixed choice agree on a single opener only over unbuffered channels; over buffered channels with free space a send commits at once.

import groovy.concurrent.*
import static groovy.concurrent.ChannelSelect.*

// Mixed offers over rendezvous channels: either send my opener or take my peer's
def ping = AsyncChannel.create()
def pong = AsyncChannel.create()
def peer = async { await ping.receive() }          // a receiver is waiting on ping
def result = await offers(send(ping, 'opener'), receive(pong)).select()
println "send committed: ${result.send}, index=${result.index}"
println "peer got: ${await peer}"
// now nobody receives on ping, but a sender is waiting on pong
async { await pong.send('peer-opener') }
Thread.sleep(100)
def result2 = await offers(send(ping, 'opener'), receive(pong)).select()
println "send committed: ${result2.send}, value=${result2.value}"

send committed: true, index=0
peer got: opener
send committed: false, value=peer-opener

Source: Groovy documentation: Async/await, Mixed offers

Guards

Any offer can carry a precondition, the guarded choice of CSP: receive(input).when { held.size() < capacity }. The guard is a closure consulted afresh on every select() call, so one select can be built once and reused while the state it guards changes. A guarded-off offer is not registered and keeps its position, so result.index always means the same branch. The same preconditions can be passed positionally as boolean flags, one per offer. Enabling no branch at all is a deadlock, so select throws IllegalStateException, and a wrong number of flags throws IllegalArgumentException.

import groovy.concurrent.*
import static groovy.concurrent.ChannelSelect.*

// Guards: a bounded buffer built from two guarded receive offers
def input = AsyncChannel.create(10)
def request = AsyncChannel.create(10)
def held = new ArrayDeque()
def capacity = 2
3.times { input.send("item-$it") }
request.send('give')
def sel = offers(receive(input).when { held.size() < capacity },
                 receive(request).when { !held.isEmpty() })
def log = []
4.times {
    def res = await sel.select()
    if (res.index == 0) { held.addLast(res.value); log << "stored ${res.value}" }
    else { log << "served ${held.removeFirst()}" }
}
println log
println "still held: $held"

// the same guards as positional flags, computed as a mask
input.send('item-3'); request.send('give')
def plain = offers(receive(input), receive(request))
def res = await plain.select(true, false)             // only the first offer enabled
println "flags (true, false): ${res.value} from index ${res.index}"
try {
    await plain.select(false, false)
} catch (IllegalStateException e) {
    println "no branch enabled: ${e.class.simpleName}"
}
try {
    await plain.select(true)
} catch (IllegalArgumentException e) {
    println "wrong flag count: ${e.class.simpleName}"
}

[stored item-0, stored item-1, served item-0, stored item-2]
still held: [item-1, item-2]
flags (true, false): item-3 from index 0
no branch enabled: IllegalStateException
wrong flag count: IllegalArgumentException

Source: Groovy documentation: Async/await, Preconditions

Timeouts

A deadline can be a branch of the select. The timer offer after(ms) (or after(Duration)) arms a fresh timer on every select() call and commits, with the Instant at which it fired, only if no channel offer committed first; its result has timeout set to true and a null channel. A select that dispatches on result.channel must therefore test result.timeout first. A positive delay is never ready at registration, and listed last, after(0) acts as Go’s default case: take a transfer if one can complete at once, otherwise carry on without waiting. The timer channel AsyncChannel.after(ms) is the fixed-deadline form: a capacity-1 channel that delivers one Instant after the delay and then closes, so it is a single deadline shared by every round that selects on it.

import groovy.concurrent.*
import static groovy.concurrent.ChannelSelect.*

// after(ms): a timer offer re-armed on every select call
def work = AsyncChannel.create(5)
def sel = offers(receive(work), after(100)).fair()
async { Thread.sleep(30); await work.send('job-1') }
def rounds = []
3.times {
    def r = await sel.select()
    rounds << (r.timeout ? "timeout (channel=${r.channel}, index=${r.index})" : "got ${r.value}")
}
println rounds

// after(0) listed last is the "default" case of a Go select
def nothing = AsyncChannel.create()
def d = await offers(receive(nothing), after(0)).select()
println "default branch: timeout=${d.timeout}, value is Instant: ${d.value instanceof java.time.Instant}"

// AsyncChannel.after: a fixed deadline shared by all rounds
def deadline = AsyncChannel.after(250)
def feed = AsyncChannel.create(10)
async { 5.times { Thread.sleep(100); await feed.send(it) } }
def sel2 = ChannelSelect.from(feed, deadline)
def got = []
while (true) {
    def r = await sel2.select()
    if (r.index == 1) break
    got << r.value
}
println "received before the deadline: $got"
println "deadline channel closed after firing: ${deadline.closed}"

[got job-1, timeout (channel=null, index=1), timeout (channel=null, index=1)]
default branch: timeout=true, value is Instant: true
received before the deadline: [0, 1]
deadline channel closed after firing: true

Source: Groovy documentation: Async/await, Timeouts

The three results of the first select show the lifecycle: the first round got job-1, and each later round re-armed its own 100 ms timer and timed out. The last loop received what arrived before the fixed deadline and then stopped, after which the timer channel reported itself closed.

BroadcastChannel and Flow.Publisher interop

An AsyncChannel is point-to-point: each value goes to one receiver. A BroadcastChannel is one-to-many: every value is delivered to all current subscribers, each of which gets its own channel from subscribe() (optionally subscribe(n) to set the buffer size). Subscribers only see messages sent after they subscribed.

asPublisher() exposes the broadcast channel as a java.util.concurrent.Flow.Publisher, so it composes with any Reactive Streams operator and with for await. Each subscription creates its own bounded channel (default 16) and honours the standard request(n) demand. The back-pressure policy is lossless and sender-gated: send waits until every live subscriber has accepted the value, so a subscriber that never requests, or requests slowly, eventually stalls the producer for everybody, and the slowest subscriber sets the throughput. For drop-oldest or latest-only behaviour, put a Reactive Streams operator that implements that policy in front of the slow subscriber. Completion is not gated by demand: once the channel is closed and a subscriber’s buffer is drained, it receives onComplete even without outstanding demand.

import groovy.concurrent.*
import java.util.concurrent.Flow

// One-to-many: every subscriber sees every value sent after it subscribed
def alarms = BroadcastChannel.create()
def dashboard = alarms.subscribe()
def pager = alarms.subscribe()
println "subscribers: ${alarms.subscriberCount}"
def sender = async {
    await alarms.send('frost')
    await alarms.send('door open')
    alarms.close()
}
// subscriber channels have a bounded buffer (16 by default), so this fits without a reader
await sender
def a = []; for await (m in dashboard) { a << m }
def b = []; for await (m in pager) { b << m }
println "dashboard=$a pager=$b"

// a late subscriber misses earlier messages
def live = BroadcastChannel.create()
def early = live.subscribe()
await live.send('m1')
def late = live.subscribe()
await live.send('m2')
live.close()
def e = []; for await (m in early) { e << m }
def l = []; for await (m in late) { l << m }
println "early=$e late=$l"

// Flow.Publisher interop: for await consumes the publisher
def bc = BroadcastChannel.create()
def publisher = bc.asPublisher()
println publisher instanceof Flow.Publisher
def consumer = async {
    def seen = []
    for await (msg in publisher) { seen << msg }
    seen
}
Thread.sleep(200)                     // let the subscription register
await bc.send('x'); await bc.send('y')
bc.close()
println "publisher consumer saw ${await consumer}"

// a hand-written Flow.Subscriber with demand
def bc2 = BroadcastChannel.create()
def got = new java.util.concurrent.CopyOnWriteArrayList()
def done = new java.util.concurrent.CountDownLatch(1)
bc2.asPublisher().subscribe(new Flow.Subscriber<String>() {
    Flow.Subscription sub
    void onSubscribe(Flow.Subscription s) { sub = s; s.request(1) }
    void onNext(String item) { got << item; sub.request(1) }
    void onError(Throwable t) { done.countDown() }
    void onComplete() { got << 'complete'; done.countDown() }
})
Thread.sleep(200)
await bc2.send('p'); await bc2.send('q'); bc2.close()
done.await()
println got

subscribers: 2
dashboard=[frost, door open] pager=[frost, door open]
early=[m1, m2] late=[m2]
true
publisher consumer saw [x, y]
[p, q, complete]

Source: Groovy documentation: Async/await, BroadcastChannel

The example waits briefly (Thread.sleep(200)) before sending so that the publisher’s subscription has registered; as the early/late line shows, anything sent before a subscription is invisible to it.

Deferred cleanup with defer

defer schedules a cleanup action to run when the enclosing async closure completes, whether it returns normally or fails. Deferred actions run in last-in, first-out order, which pairs naturally with acquisition: register the cleanup right after the resource is acquired instead of nesting try/finally blocks. If a deferred action returns an Awaitable or Future, it is awaited before the next deferred action runs, so asynchronous resources are released in order. defer is a contextual keyword, recognised only inside async closures, so an existing variable or method called defer elsewhere keeps working, as the last lines show.

import groovy.concurrent.*

def log = Collections.synchronizedList([])
def task = async {
    log << 'open connection'
    defer { log << 'close connection' }
    log << 'begin transaction'
    defer { log << 'rollback or commit' }
    log << 'save readings'
    'saved'
}
println await(task)
println log

// deferred actions run even on failure, and in LIFO order
def trail = Collections.synchronizedList([])
def failing = async {
    defer { trail << 'first registered, last run' }
    defer { trail << 'second registered, first run' }
    throw new IllegalStateException('sensor bus lost')
}
try { await failing } catch (IllegalStateException e) { trail << "caught: ${e.message}" }
println trail

// a deferred action that returns an Awaitable is awaited before the next one
def order = Collections.synchronizedList([])
def t = async {
    defer { order << 'runs last' }
    defer { async { Thread.sleep(100); order << 'slow async cleanup' } }
    'body'
}
await t
println order

// defer is only a keyword inside async closures
def defer = 'still a normal identifier'
println defer

saved
[open connection, begin transaction, save readings, rollback or commit, close connection]
[second registered, first run, first registered, last run, caught: sensor bus lost]
[slow async cleanup, runs last]
still a normal identifier

Source: Groovy documentation: Async/await, Deferred cleanup with defer

For resources in non-async code, the ordinary withCloseable/try/finally of Exceptions & Error Handling remains the right tool.

Structured concurrency with AsyncScope

Structured concurrency gives concurrent tasks a clear owner and a bounded lifetime, so no background work is orphaned and no failure disappears silently. The idea is shared by Java’s structured concurrency work, Go’s errgroup and Kotlin’s coroutine scopes (see Kotlin Coroutine Context, Cancellation & Exceptions). AsyncScope.withScope { scope -> ... } (also reachable as Awaitable.withScope) runs the block, lets it start children with scope.async { ... }, and does not return until the children are done; the value of the block is the value of withScope. On JDK 25 and later the scope is tracked with a ScopedValue, and on JDK 17 to 24 transparently with a ThreadLocal.

The defaults are fail-fast: a child’s failure cancels its siblings and propagates. The measured behaviour in 6.0.0 has some subtleties that the next script demonstrates:

If the block awaits its children, a failure surfaces immediately. The awaiting of the other children ends with a CancellationException, and the first failure reaches the caller without waiting for a sibling that is still sleeping.
If the block never awaits, the scope waits for every child to finish, even a slow one, and only then reports the failure. Cancellation marks the child cancelled; it does not interrupt a thread that is blocked in Thread.sleep or computing, so long-running children that should stop early must check for cancellation themselves.
An unchecked exception reaches the caller unchanged. A checked exception, such as IOException, is wrapped in a RuntimeException (or an UndeclaredThrowableException when it is rethrown by an await inside the block).
AsyncScope.withScope(Duration) { ... } adds a deadline for the whole scope and throws TimeoutException; it applies while the block is waiting on its children.
scope.childCount, scope.cancelAll(), AsyncScope.current() (the innermost enclosing scope, or null) and AsyncScope.create(executor) complete the API; overloads of withScope also accept a custom Executor.

import groovy.concurrent.*
import java.time.Duration
import java.util.concurrent.CancellationException
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicInteger

long since(long t0) { (System.nanoTime() - t0) / 1_000_000 }

// A scope waits for all of its children and returns the block's value
def readings = AsyncScope.withScope { scope ->
    def t1 = scope.async { Thread.sleep(80); [sensor: 'gh-1', celsius: 21.5] }
    def t2 = scope.async { Thread.sleep(40); [sensor: 'gh-2', celsius: 24.0] }
    def t3 = scope.async { [sensor: 'gh-3', celsius: 19.0] }
    [await(t1), await(t2), await(t3)]
}
println readings*.sensor
println readings.max { it.celsius }

// Children are awaited even when the block never awaits them
def completed = new AtomicInteger()
AsyncScope.withScope { scope ->
    3.times { scope.async { Thread.sleep(50); completed.incrementAndGet() } }
}
println "children completed: ${completed.get()}"

// Fail-fast: when the block awaits the children, the first failure
// cancels the siblings and propagates at once
long t0 = System.nanoTime()
try {
    AsyncScope.withScope { scope ->
        def slow = scope.async { Thread.sleep(2000); 'finally' }
        def bad = scope.async { Thread.sleep(50); throw new IllegalStateException('gh-2 bus lost') }
        await(slow, bad)
    }
} catch (IllegalStateException e) {
    println "propagated '${e.message}' after less than 1 s: ${since(t0) < 1000}"
}

// Awaiting only the slow child: it is cancelled by the sibling's failure
try {
    AsyncScope.withScope { scope ->
        def slow = scope.async { Thread.sleep(2000); 'finally' }
        scope.async { Thread.sleep(50); throw new IllegalStateException('gh-2 bus lost') }
        await slow
    }
} catch (CancellationException e) {
    println "awaited sibling was cancelled: ${e.class.simpleName}"
}

// Without any await the scope simply waits for the sleeping child before reporting the failure
t0 = System.nanoTime()
try {
    AsyncScope.withScope { scope ->
        scope.async { Thread.sleep(600); 'finally' }
        scope.async { throw new IllegalStateException('gh-2 bus lost') }
    }
} catch (IllegalStateException e) {
    println "no await: waited for the slow child (>= 500 ms): ${since(t0) >= 500}"
}

// A checked exception from a child comes out wrapped in a RuntimeException
try {
    AsyncScope.withScope { scope -> scope.async { throw new IOException('disk full') } }
} catch (RuntimeException e) {
    println "${e.class.simpleName} caused by ${e.cause.class.simpleName}: ${e.cause.message}"
}

// A deadline for the whole scope: it applies while the block waits on its children
try {
    AsyncScope.withScope(Duration.ofMillis(150)) { scope ->
        await scope.async { Thread.sleep(2000); 'too slow' }
    }
} catch (TimeoutException e) {
    println "scope timed out: ${e.class.simpleName}"
}

// Scope introspection
AsyncScope.withScope { scope ->
    scope.async { Thread.sleep(100) }
    scope.async { Thread.sleep(100) }
    println "children=${scope.childCount}, current is this scope: ${AsyncScope.current() == scope}"
}
println "outside a scope: ${AsyncScope.current()}"

// Awaitable.withScope is the same entry point
println Awaitable.withScope { s -> await s.async { 42 } }

[gh-1, gh-2, gh-3]
[sensor:gh-2, celsius:24.0]
children completed: 3
propagated 'gh-2 bus lost' after less than 1 s: true
awaited sibling was cancelled: CancellationException
no await: waited for the slow child (>= 500 ms): true
RuntimeException caused by IOException: disk full
scope timed out: TimeoutException
children=2, current is this scope: true
outside a scope: null
42

Source: Groovy documentation: Async/await, Structured concurrency

The 6.0 release notes show AsyncScope.run { ... } for the same purpose; the 6.0.0 binary used here has withScope and no run, matching the user guide. Plain async { ... } outside a scope is unstructured: nothing waits for it, so combine it with await, Awaitable.all or a scope.

Choosing the right construct

Need Use

Need	Use
a sequence of values produced over time by one task	a `yield return` generator consumed with `for await`
two or more tasks exchanging values (producer/consumer, fan-out/fan-in)	`AsyncChannel`, with `filter`/`map`/`merge`/`split`/`tap` pipelines
reacting to whichever of several sources is ready, with priorities, guards or timeouts	`ChannelSelect`
every subscriber must see every event	`BroadcastChannel`, or its `Flow.Publisher` for Reactive Streams consumers
guaranteed cleanup at the end of an async task	`defer`
child tasks that must not outlive a block, with fail-fast cancellation	`AsyncScope.withScope`
one-off results from independent tasks	`Awaitable.all`/`any`/`first`/`allSettled`; see Async/Await
data-parallel work over a collection	parallel collections; see Parallel Collections & Dataflow
long-lived stateful message handlers	actors and agents; see Actors, Agents & Active Objects

a sequence of values produced over time by one task

a yield return generator consumed with for await

two or more tasks exchanging values (producer/consumer, fan-out/fan-in)

AsyncChannel, with filter/map/merge/split/tap pipelines

reacting to whichever of several sources is ready, with priorities, guards or timeouts

ChannelSelect

every subscriber must see every event

BroadcastChannel, or its Flow.Publisher for Reactive Streams consumers

guaranteed cleanup at the end of an async task

defer

child tasks that must not outlive a block, with fail-fast cancellation

AsyncScope.withScope

one-off results from independent tasks

Awaitable.all/any/first/allSettled; see Async/Await

data-parallel work over a collection

parallel collections; see Parallel Collections & Dataflow

long-lived stateful message handlers

actors and agents; see Actors, Agents & Active Objects

Where the book is outdated

The book consulted for this section (see the section bibliography) predates all of this. It describes concurrency in terms of threads, java.util.concurrent and, at best, the external GPars library for dataflow and actors. In Groovy 6.0, generators, channels with a CSP-style select, broadcast and structured scopes are part of the distribution in groovy.concurrent, integrated with the async and await keywords and with virtual threads. Because the feature set is incubating in 6.0, verify details such as cancellation behaviour against the exact version you run, as this page does for 6.0.0.