Is parallel worth it?

One job, five ways to run it. The answer depends on exactly one number, and it is not the number people reach for.

Every "use isolates for CPU work" article stops before the part that decides it. Handing one element to another isolate and getting the result back costs about 5µs. If the work on that element costs less than that, no number of cores will save you — you have bought a courier to carry a letter across the room.

So these three cases vary one thing: the cost of a single element. Everything else — the dataset, the checksum, the worker function — is held fixed, and every program calls the same top-level function, so the only difference between them is where that function runs.

When to pass chunk, and when not to

Picture ten rooms down a hall, one person in each room. To give someone a job you walk over, hand them a sheet of paper, and walk back with the answer. That walk is about 5µs — a tiny slice of time — every time, even if the sheet is almost blank. The number 10 is the worker count: how many rooms you hired. On this page it is 10, because the machine that ran the charts has 10 cores.

Leave chunk off when the job on one sheet is heavy. Below there are 20,000 passwords and 10 workers. Rehashing one password takes about 250µs — fifty walks of work. The walk is then a rounding error. Send one password per trip. That is 20,000 walks, and that is fine: 

// 20,000 passwords, 10 workers. No chunk.
await fx(creds).parallel(10, rehash).toList();
// 20,000 trips. Each trip ~5µs, each job ~250µs.

Pass chunk when the job on one sheet is lighter than the walk. Below there are 1,500,000 log lines and 10 workers. Fingerprinting one line takes about 3.5µs — less than one walk. Send one line per trip and you pay 1,500,000 walks, which is slower than doing the work at your own desk. Instead pack 37,500 lines in each envelope (1,500,000 ~/ (10 * 4) = 37,500). Why 10 * 4? Ten rooms, four envelopes each, so 40 trips instead of 1,500,000 — and if one envelope is slower, the other rooms still have three left to share: 

// 1,500,000 log lines, 10 workers, 37,500 lines per envelope.
await fx(lines).parallel(10, fingerprint, chunk: 37500).toList();
// 40 trips. 1,500,000 / (10 * 4) = 37,500.

The five ways

  1. Native, one isolate — a plain for loop. The baseline every other row is measured against, because it is what the code looked like before anyone reached for a library.
  2. Native + dart:isolate — slice the list, one Isolate.run per slice, Future.wait, concatenate. This is what you write by hand, and it is the bar parallel has to clear: it is not enough to beat the loop.
  3. fxdart chain, one isolate — fx(xs).map(work).toList(). It shares no work with the isolates; it is here to price the chain itself, so the row below is not quietly credited with the chain's overhead or blamed for it.
  4. fxdart .parallel() — the same chain with one operator changed, at its default: every element crosses to a worker on its own. This is what you write first.
  5. fxdart .parallel(chunk:) — the same operator with k elements on each message, so the trip is paid once per batch instead of once per element. The last two rows are one row apart on purpose: the gap between them is the round trip, drawn to scale.

Reading the numbers

Each case is sized so the plain loop runs for about five seconds. That is deliberate: below a second, spawning the isolates (~1ms each) and copying the data are a large enough share of the total that the measurement is mostly about the harness. A job worth parallelising is a job that takes a while.

The two smaller blocks run the identical program at N = 10,000 and N = 100. They are not padding — they are the other half of the answer. Isolates have a fixed price: about a millisecond to spawn each one, plus copying the data in and the results back out. The smaller the job, the less is left to win, and where that crosses over is not guessable from the element count alone. Watch password-rehash still win at N = 100 while log-fingerprint has already lost at N = 10,000: it is total work that decides, not how many things there are.

What to look for. In password-rehash each element costs ~250µs — fifty times the trip — and parallel wins with no tuning at all. In log-fingerprint each element costs ~3.5µs, less than the trip, and the default parallel is slower than the plain loop. That is not a defect; it is the operator being asked to pay a per-element price for a per-element job. chunk: is the fix, and the last two rows are how much it is worth.

Why more workers cannot fix the slow row

If log-fingerprint were merely short of parallelism, a bigger pool would help. It does not. The same program at N = 100,000, varying nothing but the number of workers: 

workers   .parallel()        .parallel(chunk:)
      1     768.8 ms             381.0 ms
      2     831.9 ms             191.1 ms
      5     899.1 ms              86.5 ms
     10     873.5 ms              71.0 ms

This table is a separate measurement (BENCH_N=100000, BENCH_WORKERS 1–10). It is not in results-parallel.json, so regenerating the page charts does not refresh these four rows.

The default form does not improve at all — it drifts slightly worse, and its cost stays around 8µs per element whatever the pool size. The chunked form scales 5.4× across the same range.

That is the diagnosis. At chunk: 1 every element costs two message copies, a port event and a completer on the main isolate — which is one thread, and the one thing in the system that cannot be parallelised. About 8µs of coordination (the hop plus that completer and event) to hand off 3.5µs of work. The workers are not the bottleneck; they are idle, waiting to be fed by a main isolate that is spending all its time posting letters. Extra workers only add contention for it.

A batch does not make the coordination cheaper — it makes there be less of it. The chunked row uses n ~/ (workers * 4), so ten workers always send 40 messages. At this sweep that is chunk: 2500 instead of 100,000 round trips; at the headline (N = 1,500,000) it is chunk: 37500 instead of 1.5 million trips. The main isolate stops being the bottleneck, and the work finally lands where it was supposed to go.

Sources: benchmark/cases-parallel/. Regenerate with dart run benchmark/run_parallel_benchmarks.dart. The runner refuses a case whose variants do not all produce an identical checksum, so the rows are always different ways of computing one answer.

Measured on 10 workers, AOT (dart compile exe), median of 3 iterations (1 round(s) × 3). All 5 variants compute the same checksum — the runner refuses the case otherwise.

password-rehash

N = 20,000

Native, one isolate 4.98 s baseline
Native + dart:isolate 748.4 ms 6.65x faster
fxdart chain, one isolate 4.95 s 1.01x faster
fxdart .parallel() 794.2 ms 6.27x faster
fxdart .parallel(chunk:) 758.1 ms 6.57x faster

N = 10,000

Native, one isolate 2.48 s baseline
Native + dart:isolate 373.7 ms 6.64x faster
fxdart chain, one isolate 2.48 s same
fxdart .parallel() 417.3 ms 5.94x faster
fxdart .parallel(chunk:) 370.4 ms 6.70x faster

N = 100

Native, one isolate 25.2 ms baseline
Native + dart:isolate 5.0 ms 5.07x faster
fxdart chain, one isolate 25.2 ms same
fxdart .parallel() 4.7 ms 5.40x faster
fxdart .parallel(chunk:) 4.9 ms 5.19x faster
The five programs, and the job they share
The job itself — all five call this one function
// The per-element job, shared verbatim by all five variants.
//
// Top-level and sendable, which is what `parallel` asks of a worker and
// what `Isolate.run` needs anyway — so the four files differ only in where
// this runs, never in what it computes.

/// One credential to re-hash.
class Credential {
  const Credential(this.user, this.salt, this.secret);
  final int user;
  final int salt;
  final int secret;
}

/// The result: the derived key, and the user it belongs to.
class Derived {
  const Derived(this.user, this.key);
  final int user;
  final int key;
}

/// Iterated key derivation, PBKDF2's shape: mix the secret with the salt
/// over and over so that verifying a password is deliberately expensive.
///
/// The round count is what a KDF is *tuned* by, and it is set here so one
/// credential costs ~250 µs — the range a real deployment picks, and fifty
/// times the ~5 µs it costs to hand the credential to another isolate. That
/// ratio is the case: work this heavy does not need the batching the cheap
/// cases do.
///
/// Why not heavier still: the headline N has to stay above the 10,000 the
/// runner also measures at, or the "full" block would be *smaller* than the
/// block above it and the page would read backwards. Cost per credential and
/// the headline size trade off against each other at a fixed ~5 s baseline.
const kdfRounds = 55000;

Derived rehash(Credential c) {
  var h = c.secret ^ (c.salt * 0x9E3779B1);
  for (var i = 0; i < kdfRounds; i++) {
    h = (h * 31 + c.salt + i) & 0x1FFFFFFFFFFFFF;
    h ^= (h >> 13);
    h = (h * 0x27D4EB2D) & 0x1FFFFFFFFFFFFF;
    h ^= (h >> 7);
  }
  return Derived(c.user, h);
}

/// The same job over a slice, for the hand-rolled isolate variant.
List<Derived> rehashAll(List<Credential> batch) => [
  for (final c in batch) rehash(c),
];
Native, one isolate
// 1 of 5 — a plain loop. One isolate, no chain. The baseline.
import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final creds = makeCredentials();
  await bench(
    slug: 'password-rehash',
    impl: 'native',
    n: n,
    run: () {
      final out = <Derived>[];
      for (final c in creds) {
        out.add(rehash(c));
      }
      return checksum(out);
    },
  );
}
Native + dart:isolate
// 2 of 5 — hand-rolled isolates: slice the list, one Isolate.run per slice,
// wait for all, concatenate. This is what you write when you reach for
// dart:isolate directly, and it is what `parallel` has to beat.
import 'dart:isolate';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final creds = makeCredentials();
  await bench(
    slug: 'password-rehash',
    impl: 'native-isolate',
    n: n,
    run: () async {
      final slices = sliceEvenly(creds, benchWorkers);
      final parts = await Future.wait([
        for (final s in slices) Isolate.run(() => rehashAll(s)),
      ]);
      return checksum([for (final p in parts) ...p]);
    },
  );
}
fxdart chain, one isolate
// 3 of 5 — the fxdart chain, still on one isolate. Isolates the cost of the
// chain itself, so the parallel row below is not credited with it.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final creds = makeCredentials();
  await bench(
    slug: 'password-rehash',
    impl: 'fxdart',
    n: n,
    run: () => checksum(fx(creds).map(rehash).toList()),
  );
}
fxdart .parallel()
// 4 of 5 — the same chain, one operator changed. The default form: every
// element crosses to a worker on its own.
//
// ~250 µs per credential dwarfs the ~5 µs round trip, so streaming is
// already the right call here — the trip is noise against the work, and
// every worker stays busy without waiting for a batch to fill.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final creds = makeCredentials();
  await bench(
    slug: 'password-rehash',
    impl: 'fxdart-parallel',
    n: n,
    run: () async =>
        checksum(await fx(creds).parallel(benchWorkers, rehash).toList()),
  );
}
fxdart .parallel(chunk:)
// 5 of 5 — the same operator, with `chunk` set.
//
// `chunk: k` puts k elements on one message instead of one each, so the
// round trip is paid once per batch. `length ~/ (workers * 16)` leaves every
// worker 16 turns, which is enough to balance uneven elements without
// paying per element.
//
// ~250 us per credential against a ~5 us round trip: the trip is already
// noise, so the default streaming form is the right one and a batch has
// almost nothing left to save. This row is here to show that.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final creds = makeCredentials();
  final chunk = (n ~/ (benchWorkers * 16)).clamp(1, 1 << 30);
  await bench(
    slug: 'password-rehash',
    impl: 'fxdart-parallel-chunk',
    n: n,
    run: () async => checksum(
      await fx(creds).parallel(benchWorkers, rehash, chunk: chunk).toList(),
    ),
  );
}

image-tiles

N = 147,000

Native, one isolate 5.56 s baseline
Native + dart:isolate 1.68 s 3.32x faster
fxdart chain, one isolate 5.50 s 1.01x faster
fxdart .parallel() 1.75 s 3.17x faster
fxdart .parallel(chunk:) 781.9 ms 7.11x faster

N = 10,000

Native, one isolate 326.3 ms baseline
Native + dart:isolate 87.6 ms 3.73x faster
fxdart chain, one isolate 317.7 ms 1.03x faster
fxdart .parallel() 117.9 ms 2.77x faster
fxdart .parallel(chunk:) 56.0 ms 5.82x faster

N = 100

Native, one isolate 3.3 ms baseline
Native + dart:isolate 0.9 ms 3.82x faster
fxdart chain, one isolate 3.2 ms 1.05x faster
fxdart .parallel() 1.4 ms 2.46x faster
fxdart .parallel(chunk:) 1.4 ms 2.38x faster
The five programs, and the job they share
The job itself — all five call this one function
// The per-element job, shared verbatim by all five variants.
//
// The middle of the three: ~40 µs per tile, so the round trip is a tenth of
// the work rather than all of it or none of it.

import 'dart:typed_data';

/// One 32×32 greyscale tile of a larger image.
class Tile {
  const Tile(this.index, this.pixels);
  final int index;
  final Uint8List pixels;
}

/// What the filter produced: the tile's index and its summary statistics.
class TileStats {
  const TileStats(this.index, this.edgeEnergy, this.mean);
  final int index;
  final int edgeEnergy;
  final int mean;
}

const tileSide = 32;

/// How many filter passes each tile gets. A sharpen stage runs a stack of
/// them in any real pipeline; eight puts one tile at ~42 µs, which is the
/// middle of the three cases — several times the ~5 µs isolate round trip,
/// but not the 100x that [password-rehash] has.
const passes = 8;

/// A 3×3 Sobel pass over the tile, reduced to two numbers. Real image work:
/// every output pixel reads nine inputs, so it is memory-bound in a way a
/// synthetic spin loop is not.
TileStats sharpen(Tile tile) {
  final p = tile.pixels;
  var energy = 0;
  var sum = 0;
  for (var pass = 0; pass < passes; pass++) {
    (energy, sum) = _pass(p, energy, sum);
  }
  final inner = (tileSide - 2) * (tileSide - 2);
  return TileStats(tile.index, energy, sum ~/ (inner * passes));
}

(int, int) _pass(Uint8List p, int energy, int sum) {
  for (var y = 1; y < tileSide - 1; y++) {
    final row = y * tileSide;
    for (var x = 1; x < tileSide - 1; x++) {
      final i = row + x;
      final gx =
          -p[i - tileSide - 1] +
          p[i - tileSide + 1] -
          2 * p[i - 1] +
          2 * p[i + 1] -
          p[i + tileSide - 1] +
          p[i + tileSide + 1];
      final gy =
          -p[i - tileSide - 1] -
          2 * p[i - tileSide] -
          p[i - tileSide + 1] +
          p[i + tileSide - 1] +
          2 * p[i + tileSide] +
          p[i + tileSide + 1];
      energy += (gx < 0 ? -gx : gx) + (gy < 0 ? -gy : gy);
      sum += p[i];
    }
  }
  return (energy, sum);
}

/// The same job over a slice, for the hand-rolled isolate variant.
List<TileStats> sharpenAll(List<Tile> batch) => [
  for (final t in batch) sharpen(t),
];
Native, one isolate
// 1 of 5 — a plain loop. One isolate, no chain. The baseline.
import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final tiles = makeTiles();
  await bench(
    slug: 'image-tiles',
    impl: 'native',
    n: n,
    run: () {
      final out = <TileStats>[];
      for (final t in tiles) {
        out.add(sharpen(t));
      }
      return checksum(out);
    },
  );
}
Native + dart:isolate
// 2 of 5 — hand-rolled isolates: one Isolate.run per slice.
import 'dart:isolate';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final tiles = makeTiles();
  await bench(
    slug: 'image-tiles',
    impl: 'native-isolate',
    n: n,
    run: () async {
      final slices = sliceEvenly(tiles, benchWorkers);
      final parts = await Future.wait([
        for (final s in slices) Isolate.run(() => sharpenAll(s)),
      ]);
      return checksum([for (final p in parts) ...p]);
    },
  );
}
fxdart chain, one isolate
// 3 of 5 — the fxdart chain, still on one isolate.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final tiles = makeTiles();
  await bench(
    slug: 'image-tiles',
    impl: 'fxdart',
    n: n,
    run: () => checksum(fx(tiles).map(sharpen).toList()),
  );
}
fxdart .parallel()
// 4 of 5 — the same chain, one operator changed. The default form: every
// element crosses to a worker on its own.
//
// ~37 µs per tile against a ~5 µs round trip. The trip is about a tenth of
// the work: enough to notice, not enough to lose to.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final tiles = makeTiles();
  await bench(
    slug: 'image-tiles',
    impl: 'fxdart-parallel',
    n: n,
    run: () async => checksum(
      await fx(tiles).parallel(benchWorkers, sharpen).toList(),
    ),
  );
}
fxdart .parallel(chunk:)
// 5 of 5 — the same operator, with `chunk` set.
//
// `chunk: k` puts k elements on one message instead of one each, so the
// round trip is paid once per batch. `length ~/ (workers * 16)` leaves every
// worker 16 turns, which is enough to balance uneven elements without
// paying per element.
//
// ~37 us per tile against a ~5 us round trip: the trip is about a tenth of
// the work, so batching has something to take but the default already wins.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final tiles = makeTiles();
  final chunk = (n ~/ (benchWorkers * 16)).clamp(1, 1 << 30);
  await bench(
    slug: 'image-tiles',
    impl: 'fxdart-parallel-chunk',
    n: n,
    run: () async => checksum(
      await fx(tiles).parallel(benchWorkers, sharpen, chunk: chunk).toList(),
    ),
  );
}

log-fingerprint

N = 1,500,000

Native, one isolate 5.44 s baseline
Native + dart:isolate 3.81 s 1.43x faster
fxdart chain, one isolate 5.43 s same
fxdart .parallel() 13.68 s 2.52x slower
fxdart .parallel(chunk:) 1.16 s 4.69x faster

N = 10,000

Native, one isolate 34.8 ms baseline
Native + dart:isolate 9.0 ms 3.87x faster
fxdart chain, one isolate 34.7 ms same
fxdart .parallel() 87.4 ms 2.51x slower
fxdart .parallel(chunk:) 6.7 ms 5.17x faster

N = 100

Native, one isolate 0.4 ms baseline
Native + dart:isolate 0.3 ms 1.17x faster
fxdart chain, one isolate 0.3 ms 1.03x faster
fxdart .parallel() 1.2 ms 3.53x slower
fxdart .parallel(chunk:) 0.9 ms 2.55x slower
The five programs, and the job they share
The job itself — all five call this one function
// The per-element job, shared verbatim by all five variants.
//
// Deliberately cheap — a few microseconds — because that is the case where
// the isolate round trip costs more than the work, and where `chunk` stops
// being a tuning knob and becomes the whole difference.

/// One raw log line.
class LogLine {
  const LogLine(this.id, this.text);
  final int id;
  final String text;
}

/// A line reduced to the shape it shares with every other line like it.
class Fingerprint {
  const Fingerprint(this.id, this.hash, this.digits);
  final int id;
  final int hash;
  final int digits;
}

/// How many hash permutations the sketch keeps. This is the knob the case
/// is calibrated on: it sets the per-line cost, and the point of this case
/// is that the cost lands *below* the ~5 µs it takes to hand one line to
/// another isolate. Real MinHash sketches run 16-128 permutations.
const sketchSize = 96;

/// Shingle width — overlapping character n-grams, so a line that differs by
/// one token still shares most of its shingles with the lines like it.
const shingle = 5;

/// Normalise a log line and reduce it to a MinHash sketch.
///
/// Digit runs collapse to `#` first, so ids and durations do not make every
/// line unique; then the normalised text is shingled and each shingle is fed
/// through [sketchSize] cheap permutations, keeping the minimum of each. Two
/// lines of the same shape land on the same sketch, which is how a log
/// pipeline groups a million lines into a handful of templates.
///
/// ~3 µs per line — *less* than the ~5 µs round trip to an isolate. That is
/// this case: at `chunk: 1`, `parallel` loses to the plain loop no matter
/// how many workers it is given, because the trip costs more than the trip
/// is for. It is the one case where `chunk` is not a tuning knob but the
/// difference between winning and losing.
Fingerprint fingerprint(LogLine line) {
  final text = line.text;
  var digits = 0;

  // Normalise in place into a small code-unit buffer: digit runs to one `#`.
  final norm = List<int>.filled(text.length, 0);
  var len = 0;
  var lastWasDigit = false;
  for (var i = 0; i < text.length; i++) {
    final c = text.codeUnitAt(i);
    final isDigit = c >= 0x30 && c <= 0x39;
    if (isDigit) {
      digits++;
      if (lastWasDigit) continue;
      norm[len++] = 0x23;
    } else {
      norm[len++] = c;
    }
    lastWasDigit = isDigit;
  }

  var sketch = 0x7FFFFFFF;
  var mixed = 0;
  for (var start = 0; start + shingle <= len; start++) {
    // One rolling hash per shingle...
    var h = 0x811C9DC5;
    for (var k = 0; k < shingle; k++) {
      h = ((h ^ norm[start + k]) * 0x01000193) & 0x3FFFFFFF;
    }
    // ...then the permutations, keeping each one's running minimum. The
    // minima are folded together rather than kept as a vector: the case
    // needs the *cost* of a sketch, not the sketch itself.
    for (var p = 0; p < sketchSize; p++) {
      final v = ((h + p * 0x9E3779B1) * 0x85EBCA6B) & 0x3FFFFFFF;
      if (v < sketch) sketch = v;
      mixed = (mixed + (v & 0x3F)) & 0x3FFFFFFF;
    }
  }
  return Fingerprint(line.id, (sketch * 31 + mixed) & 0x3FFFFFFF, digits);
}

/// The same job over a slice, for the hand-rolled isolate variant.
List<Fingerprint> fingerprintAll(List<LogLine> batch) => [
  for (final l in batch) fingerprint(l),
];
Native, one isolate
// 1 of 5 — a plain loop. One isolate, no chain. The baseline.
import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final lines = makeLines();
  await bench(
    slug: 'log-fingerprint',
    impl: 'native',
    n: n,
    run: () {
      final out = <Fingerprint>[];
      for (final l in lines) {
        out.add(fingerprint(l));
      }
      return checksum(out);
    },
  );
}
Native + dart:isolate
// 2 of 5 — hand-rolled isolates: one Isolate.run per slice.
//
// Note what this variant does *not* pay: the slices are sent once, so it is
// already the batched shape. That is exactly why plain `parallel` loses here
// and `parallel(chunk:)` does not — the comparison is only fair once both
// sides send the same number of messages.
import 'dart:isolate';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final lines = makeLines();
  await bench(
    slug: 'log-fingerprint',
    impl: 'native-isolate',
    n: n,
    run: () async {
      final slices = sliceEvenly(lines, benchWorkers);
      final parts = await Future.wait([
        for (final s in slices) Isolate.run(() => fingerprintAll(s)),
      ]);
      return checksum([for (final p in parts) ...p]);
    },
  );
}
fxdart chain, one isolate
// 3 of 5 — the fxdart chain, still on one isolate.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final lines = makeLines();
  await bench(
    slug: 'log-fingerprint',
    impl: 'fxdart',
    n: n,
    run: () => checksum(fx(lines).map(fingerprint).toList()),
  );
}
fxdart .parallel()
// 4 of 5 — the same chain, one operator changed. The default form: every
// element crosses to a worker on its own.
//
// ~3.5 µs of work against a ~5 µs round trip — the trip costs more than the
// trip is for, and no number of workers fixes that. This row is expected to
// lose to the plain loop, and the next one is why that is a tuning problem
// rather than a verdict on the operator.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final lines = makeLines();
  await bench(
    slug: 'log-fingerprint',
    impl: 'fxdart-parallel',
    n: n,
    run: () async => checksum(
      await fx(lines).parallel(benchWorkers, fingerprint).toList(),
    ),
  );
}
fxdart .parallel(chunk:)
// 5 of 5 — the same operator, with `chunk` set.
//
// `chunk: k` puts k elements on one message instead of one each, so the
// round trip is paid once per batch. `length ~/ (workers * 4)` leaves every
// worker 4 turns, which is enough to balance uneven elements without
// paying per element.
//
// ~3.5 us per line against a ~5 us round trip. This is the row the page is
// about: the batch is the difference between losing to a plain loop and
// beating it several times over.
import 'package:fxdart/fxdart.dart';

import '../../harness.dart';
import 'data.dart';
import 'work.dart';

Future<void> main() async {
  final lines = makeLines();
  final chunk = (n ~/ (benchWorkers * 4)).clamp(1, 1 << 30);
  await bench(
    slug: 'log-fingerprint',
    impl: 'fxdart-parallel-chunk',
    n: n,
    run: () async => checksum(
      await fx(lines).parallel(benchWorkers, fingerprint, chunk: chunk).toList(),
    ),
  );
}