¿Merece la pena parallel?

// 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);
    },
  );
}

// 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]);
    },
  );
}

// 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()),
  );
}

// 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()),
  );
}

// 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(),
    ),
  );
}

// 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);
    },
  );
}

// 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]);
    },
  );
}

// 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()),
  );
}

// 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(),
    ),
  );
}

// 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(),
    ),
  );
}

// 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);
    },
  );
}

// 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]);
    },
  );
}

// 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()),
  );
}

// 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(),
    ),
  );
}

// 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(),
    ),
  );
}