本页尚未翻译，因此以英文显示。 参与翻译

FxDart 入门 › 章节 11 · 异步 & 并发 › parallel

parallel

Overlaps CPU work across isolates, in source order. Not concurrent with a different name.

Lecture

concurrent(n) overlaps Futures on the same isolate — I/O. Dart's CPU-bound story is isolates. parallel(n, worker) is the twin: a reused pool of n isolates, results in source order, like mapConcurrent is the combined form of map-plus-concurrent. They are not the same operator — the comparison lives on concurrent or parallel.

Prefer a top-level or static function. A closure that captures a non-sendable (a ReceivePort, an open socket) throws ArgumentError at spawn — the isolate contract, not a fxdart invention. An unsendable input or result fails that pull the same way, rather than hanging. On the web the operator throws UnsupportedError — use concurrent(n) there. This listing is VM-only and is not a live playground. The worker may return a Future (FutureOr, same shape as mapConcurrent) — a sync callback is still the fast path. Nested parallel inside an async worker is allowed: that isolate spawns its own pool, and cancel of the outer chain shuts the nested pool down. One level of nesting is the contract — a third nested parallel is killed with its parent, so it cannot shut down its children.

Don't want to pick n? parallelWorkers is the VM's processor count — pass it as the first argument. A List shorter than n sizes the pool to the list, so parallel(8, w) over two items starts two isolates, not eight. People coming from mapConcurrent can write mapParallel; it is the same operator.

int timesTen(int x) => x * 10;

Future<void> main() async {
  print(await fx([1, 2, 3, 4]).parallel(2, timesTen).toList());
  // [10, 20, 30, 40]
}

chunk — how many elements ride one message

By default every element crosses to a worker on its own. That round trip costs about 5µs, which is more than most callbacks cost, and it is the whole reason a cheap worker is slower under parallel than in a plain loop. chunk: k pays it once per k elements:

// 20,000 elements, ~0.4µs of work each, 4 workers:
await fx(rows).parallel(4, parseRow).toList();             // ~142ms
await fx(rows).parallel(4, parseRow, chunk: 512).toList(); //   ~3ms

// the same work in a plain loop, no isolates:            //   ~8ms

47× on that shape, and the batched form is the first one that actually beats the loop it replaced. Size k so that k × callback is comfortably more than 5µs, while still leaving several batches per worker to balance across — length ~/ (workers * 4) is a fine starting point.

A batch does not change what you observe: order is the same, back-pressure is the same, and a worker that throws still emits the results of the elements before it and then raises on the element that actually failed. Two things do change. The first element now waits for its whole batch, so a take(1) wants a small chunk or none. And an unsendable input or result fails its whole batch rather than only its own pull — finding which element was at fault would mean sending them separately, which is the cost the batch exists to avoid.

Don't want to write chunk: n ~/ (workers * 4) and repeat the worker count? chunked: true does that from the source length:

await fx(rows).parallel(4, parseRow, chunked: true);
// k = rows.length ~/ 16 — one 4, not two

The source must be a List. A generator or an async source has no length — pass chunk: k instead. chunk: and chunked: together throw; the call has one policy.

Two CPU stages, one hop

Two .parallel calls copy every result back to this isolate and out again. Compose the workers with fxPipe2 so both stages run on the worker:

await fx(blobs)
    .parallel(4, fxPipe2(decodePng, thumbnail), chunk: 64)
    .toList();

decodePng and thumbnail must be sendable, same as any parallel worker. The returned function captures both. Add .then for more stages — no arity cap. The last .then is the worker.

Reuse the pool

parallel spawns on the first pull and kills the isolates when that chain ends. Two jobs then pay startup twice. IsolatePool is the spawn-once bracket. IsolatePool.using kills in finally, even if the body throws. Cancel of one parallelOn chain does not kill the pool — the next chain can use it.

await IsolatePool.using(4, (pool) async {
  final a = await fx(batchA).parallelOn(pool, parseRow, chunk: 256).toList();
  final b = await fx(batchB).parallelOn(pool, parseRow, chunk: 256).toList();
  return (a, b);
});