go_channels #

Go-style concurrency for Dart: typed channels, a faithful select over many channel operations, and structured task scopes with cooperative cancellation.

Dart's dart:async gives you futures and streams, but not the primitives Go and Kotlin developers reach for: a typed channel, a select that waits on several operations at once, or a scope where one task's failure cancels its siblings. go_channels adds those, in plain Dart, on a single isolate.

import 'package:go_channels/go_channels.dart';

final ch = Channel<int>(capacity: 1);
await ch.send(1);
print(await ch.receive()); // 1

Why this instead of what you already have #

Instead of Future.any. Racing two channel receives is what dart:async already offers, and it cannot withdraw the loser. Future.any attaches a callback to every future in the list and returns the first result (future.dart:645-646, Dart 3.11); the losing b.receive() stays registered. Send to b afterwards and that abandoned receive takes the value: b.length is 0 and nothing is awaiting what it took. A select withdraws instead. A send branch commits only when it wins the shared claim (channel.dart:193), and a branch that loses is removed from the queue by the closure _addSelectSend returns (channel.dart:238).

Instead of cross_channel. Its receive branch withdraws correctly: recvCancelable hands back a cancel() that removes the pop waiter (ops.dart:224, 261). Its send branch does not. select.dart:479-482 passes sender.send(value) into onFuture as an argument, so the send runs before the race begins, and send pushes on its fast path with no await above it (ops.dart:74). On 0.12.0, a select whose timeout branch wins still leaves the value in the channel.

Reach for it when #

• You are porting a Go service and want select to keep its meaning, including on the send side.
• A producer must not commit a value when a timeout or cancellation branch wins the race.
• One task's failure should cancel its siblings without threading a Completer through every call.

Skip it if a single Stream and its StreamSubscription already model the problem. That is the smaller tool, it is built in, and most Dart code does not need a second concurrency vocabulary.

Channels #

A Channel<T> passes values between asynchronous tasks. Unbuffered channels (the default) rendezvous: a send completes only when a receiver takes the value. Buffered channels hold up to capacity values before a send blocks.

The whole distinction is when send completes, which is what decides whether the sender feels backpressure the instant the receiver stalls:

final jobs = Channel<String>();              // unbuffered
final queue = Channel<String>(capacity: 32); // buffered

await for (final job in jobs.stream) {        // ranges until the channel closes
  handle(job);
}

Closing a channel lets receivers drain what is left, then observe closure. receiveOr mirrors Go's v, ok := <-ch:

final (value, ok) = await ch.receiveOr();
if (!ok) print('channel closed');

select #

select waits on several channel operations and runs exactly one. It then withdraws the branches that lost. That second half is the whole point, and it is what you cannot write with dart:async.

Race two receives the way Dart invites you to, and both of them pull a value out of their channel before Future.any ever looks. The loser's value is handed to a future nobody awaits, and it is gone:

// Silently loses a value. Both receives run; only one result is ever read.
final first = await Future.any([a.receive(), b.receive()]);

Nothing throws, no test fails, and both producers see a successful send. No signal anywhere says that a value reached nobody. A caller cannot fix this either: Dart offers no way to cancel a pending receive, which leaves the withdrawal to the channel itself. That is what select does:

final label = await select<String>((s) {
  s.onReceive(jobs, (job, ok) => ok ? 'job: $job' : 'jobs closed');
  s.onSend(results, 42, () => 'sent a result');
  s.onTimeout(const Duration(seconds: 1), () => 'timed out');
});

example/select_multiway.dart measures both spellings back to back. One run, same winner both times. The only thing that differs is whether the losing channel kept its value:

won: a=1  ->  a: empty, b: still holds 2      # select: the 2 is still there
Future.any won: a=1  ->  a: empty, b: empty   # Future.any: the 2 is destroyed

The withdrawal also matters when nothing wins, which is the case a worker waiting on work-or-shutdown spends all day in. A select that times out parks a waiter on every channel it was watching and takes all of them back out again, leaving the loop flat; the Future.any spelling abandons two receivers per round and the channel holds them for as long as it lives. Five rounds of each, from the same example:

after 5 select rounds:     a.waiters=0, b.waiters=0
after 5 Future.any rounds: c.waiters=5, d.waiters=5   # one per round, each

If more than one branch is ready at once, select picks one at random, matching Go's fairness. A busy channel cannot starve a quiet one. Declaration order is not priority: the same run split 2000 ties 973 to 1027.

Add onDefault to make the whole select non-blocking:

await select<void>((s) {
  s.onReceive(events, (e, ok) => handle(e));
  s.onDefault(() {}); // returns immediately if nothing is ready
});

Structured task scopes #

withTaskScope runs a group of tasks and returns only once all of them finish. If any task fails, the scope's token is cancelled so the siblings can stop, and the first error is rethrown. Nothing spawned inside outlives the scope.

final results = await withTaskScope((scope) async {
  final a = scope.spawn((_) => fetchA());
  final b = scope.spawn((_) => fetchB());
  return [await a, await b];
});

waitAll is the common case in one call:

final pages = await waitAll([
  (_) => fetch('/a'),
  (_) => fetch('/b'),
  (_) => fetch('/c'),
]);

Cancellation #

Dart futures cannot be forcibly killed, so cancellation is cooperative: a task observes its CancelToken and stops itself. Check isCancelled, call throwIfCancelled, or await whenCancelled inside a select.

await withTimeout(const Duration(seconds: 5), (token) async {
  while (!token.isCancelled) {
    await doOneChunk();
  }
});

example/cancellation.dart counts the steps each task got done: a failing task cutting its siblings short, a deadline the second task never reads and outlives, and a worker parked in select waiting on work-or-shutdown.

A note on parallelism #

go_channels coordinates asynchronous tasks on one isolate. It does not add parallelism by itself: use it to structure concurrent work, and combine it with isolates when you need more than one core.

Choosing a capacity #

The obvious reason to buffer a channel is speed, and on one isolate that turns out not to be the reason. Pushing 200,000 values from one task to another, every buffered size lands within a few percent of every other, about 10% above the unbuffered rendezvous:

A waiting rendezvous does not block the isolate. With a send outstanding and no receiver in sight, a 1 ms periodic timer still fired 20 times in 20 ms. There is no stalled thread for a buffer to buy back the way there would be with OS threads.

What capacity actually decides is when a producer feels backpressure: unbuffered, send waits for a receiver, so a slow consumer throttles the producer immediately; buffered, the producer runs ahead until the buffer fills. Pick it for that, not for throughput.

Draining through select is not a slow path either. Measured on the same channel it came out slightly faster than a direct receive; a worker that waits on work-or-shutdown pays nothing for the extra branch.

Numbers from benchmark/capacity_benchmark.dart on an Apple M-series core, stable across runs; benchmark/throughput_benchmark.dart measures the raw round-trip rate separately.

Dart is landing shared-memory multithreading (Isolate.runShared, tracked in dart-lang/sdk#56841). As that stabilizes, go_channels will offer a shared-memory execution path behind a capability check, and the same channel and select code will run across threads unchanged.

Status #

Stable at 1.0.0. The surface is small and follows Go's, which makes reshaping unlikely; every public type is final, which leaves room to add to the package (the shared-memory path above, for one) without breaking callers. Issues and feedback are welcome.