Dart の非同期・Stream・Isolate 実測：354ms の停止を 18ms に

シリーズ：Dart 講座 5/6（上級）。 Dart や Flutter はもう書いていて、「JSON をパースしている間に画面が固まる」「try/catch を書いたのに捕まらない」「Bad state: Stream has already been listened to」のどれかに出会ったことがある人向けです。読み終えると、非同期コードの実行順を予測し、独立した呼び出しを並列にし、Stream を正しくキャンセルし、CPU 処理を得になるときだけ isolate に移せるようになります。

Flutter の現場でよく聞くのは「async を付けたのにスクロールがカクつく」という相談です。答えは短く、Dart は isolate 1 つにスレッド 1 本で、async はコードがいつ動くかを変えるだけで、どこで動くかは変えません。以下の主張はすべて実際に動かした小さなプログラムとその出力で裏付けており、書いている途中で想定外に踏んだエラー 2 つもそのまま載せています。

実行環境： Dart SDK 3.11.5（stable）、特記なければ dart run（JIT）、Apple Silicon Mac（M1 Pro、8 コア、16GB）、macOS。時間は 3 回の中央値です。Flutter の compute() はアプリを動かしたのではなく、Flutter 3.41.9 のソースを読んで確認しました。

イベントループ：マイクロタスクが常に先

isolate にはスレッド 1 本とキュー 2 つがあります。イベントキューには外から来る仕事（タイマー、I/O の完了、Future(() {...})）、マイクロタスクキューには短い後続処理（scheduleMicrotask、完了済み Future の then コールバック、await の後ろのコード）が入ります。ルールは単純で厳格です。同期コードが終わると、ループはマイクロタスクキューをすべて空にしてから、次のイベントを 1 つ取り出します。

import 'dart:async';

void main() {
  print('A main start');
  Timer.run(() => print('B Timer.run'));
  Future(() => print('C Future(() {...})'));
  Future.delayed(Duration.zero, () => print('D Future.delayed(zero)'));
  scheduleMicrotask(() => print('E scheduleMicrotask'));
  Future.value(1).then((_) => print('F Future.value(1).then'));
  Future.microtask(() {
    print('G Future.microtask');
    scheduleMicrotask(() => print('H microtask queued by a microtask'));
  });
  () async {
    print('I async body, before the first await');
    await null;
    print('J async body, after await null');
  }();
  print('K main end');
}

A main start
I async body, before the first await
K main end
E scheduleMicrotask
F Future.value(1).then
G Future.microtask
J async body, after await null
H microtask queued by a microtask
B Timer.run
C Future(() {...})
D Future.delayed(zero)

意外な点が 3 つあります。I が K より先に出ます。async 関数は最初の await までは同期的に実行されるからです。マイクロタスクが追加したマイクロタスク H も、先に予約したタイマー B を追い越します。そして Future.delayed(Duration.zero) は「今すぐ」ではなく「すでに並んでいるものが全部終わった後」です。

Flutter に置き換えると、フレームはメイン isolate がイベントループに戻って初めて描けます。長いマイクロタスクの連鎖や長い同期関数は、周りに await がいくつあってもフレームを引き止めます。

Future・async/await・Future.wait を実測

Future<T> は後で届く値（またはエラー）で、await は届くまで関数を止めます。ありがちな失敗は、互いに無関係な呼び出しを 1 つずつ待つことです。起動画面で設定、ユーザー情報、広告設定を順に読む処理を考えてください。

Future<String> fetch(String name, int ms) async {
  await Future.delayed(Duration(milliseconds: ms));
  return '$name($ms ms)';
}

Future<void> main() async {
  final sw = Stopwatch()..start();
  final a = await fetch('config', 300);
  final b = await fetch('user', 200);
  final c = await fetch('ads', 100);
  print('sequential: $a $b $c -> ${sw.elapsedMilliseconds} ms');

  sw.reset();
  final r = await Future.wait([fetch('config', 300), fetch('user', 200), fetch('ads', 100)]);
  print('Future.wait: $r -> ${sw.elapsedMilliseconds} ms');

  sw.reset();
  final (x, y, z) = await (fetch('config', 300), fetch('user', 200), fetch('ads', 100)).wait;
  print('record .wait: $x $y $z -> ${sw.elapsedMilliseconds} ms');
}

sequential: config(300 ms) user(200 ms) ads(100 ms) -> 610 ms
Future.wait: [config(300 ms), user(200 ms), ads(100 ms)] -> 305 ms
record .wait: config(300 ms) user(200 ms) ads(100 ms) -> 303 ms

3 回の結果は、順番に待つと 610 / 615 / 611ms、Future.wait で 305 / 302 / 303ms。並列にすると合計ではなく一番遅い呼び出しの時間になります。レコード形式の (f1, f2, f3).wait は要素ごとの型（String、int …）を保てる点で、List<T> を返す Future.wait と違います。

あまり知られていないのは失敗時の挙動です。 50ms の呼び出しが例外を投げるようにしました。

Future.wait failed after 303 ms: Bad state: ads down
eagerError: true failed after 56 ms: Bad state: ads down
record .wait failed after 302 ms: ParallelWaitError<(String?, String?), (AsyncError?, AsyncError?)>

既定の Future.wait は Future がすべて終わるまで待ってから最初のエラーを報告します。早く失敗しても一番遅い呼び出しの分だけ待たされるわけです。eagerError: true を渡すとすぐに失敗します（残りの Future は動き続け、結果が捨てられるだけ）。レコードの .wait は成功した値とエラーの両方を持つ ParallelWaitError を投げるので、取れたものだけでも画面に出したいときに便利です。

よくある間違い：try/catch をすり抜ける await 忘れ

間違い： try の中で async 関数を await なしで呼ぶ。

Future<void> saveScore(int score) async {
  await Future.delayed(const Duration(milliseconds: 10));
  throw Exception('disk full while saving $score');
}

Future<void> main() async {
  try {
    saveScore(42); // no await
    print('try block finished');
  } catch (e) {
    print('caught: $e'); // never runs
  }
  await Future.delayed(const Duration(milliseconds: 50));
  print('main done'); // never reached either
}

実際に起きること：

try block finished
Unhandled exception:
Exception: disk full while saving 42
#0      saveScore (~/demo/unawaited.dart:3:3)
<asynchronous suspension>

プロセスは終了コード 255 で終わり、main done は出ませんでした。try ブロックはエラーが生まれる 10ms 前に終わっていたので、受け止める場所がなかったのです。Flutter アプリなら同じエラーは PlatformDispatcher.onError やゾーンのハンドラへ行き、リリースビルドのユーザーには「保存されていない」という結果だけが残ります。

直し方： await saveScore(42); と書けば、出力は caught: Exception: disk full while saving 42、続いて main done になります。ツールに探させるには analysis_options.yaml でリントを有効にします。

linter:
  rules:
    - unawaited_futures

   info - unawaited.dart:8:5 - Missing an 'await' for the 'Future' computed by this expression. Try adding an 'await' or wrapping the expression with 'unawaited'. - unawaited_futures

あえて待たない場合（アクセス解析の送信など）は dart:async の unawaited(...) で包み、.catchError を付けて、意図とエラー処理の両方をコードに残します。

Stream：単一購読とブロードキャスト、そしてキャンセル

Stream<T> は時間をかけて複数の値を届けます。たいていの Stream（async* 関数、ファイル読み込み、HTTP の本文）は単一購読で、リスナーは一生に 1 つだけです。

Stream<int> ticks(int n) async* {
  for (var i = 1; i <= n; i++) {
    await Future.delayed(const Duration(milliseconds: 20));
    yield i;
  }
}
// ...
final s = ticks(3);
s.listen((v) => print('first listener: $v'));
s.listen((v) => print('second listener: $v')); // throws

ERROR StateError: Bad state: Stream has already been listened to.
first listener: 1
first listener: 2
first listener: 3

Flutter では、build() の中で作った Stream を StreamBuilder に渡して再ビルドされたときや、2 つのウィジェットで 1 つの Stream を共有したときによく出ます。asBroadcastStream() や StreamController.broadcast() ならリスナーを複数持てます（実行すると A got 1、B got 1、A got 2 … の順）。ただしブロードキャスト Stream はバッファしません。誰も聞いていないときに流したイベントは消えます。

final bus = StreamController<String>.broadcast();
bus.add('login');            // nobody listening yet
await Future.delayed(Duration.zero);
bus.stream.listen((e) => print('late listener got: $e'));
bus.add('purchase');

late listener got: purchase

login は何の警告もなく捨てられました。後から来たリスナーにも最新値が必要なら、値をフィールドに持っておき、購読時に流します。

キャンセルは受信をやめるだけでなく、仕事そのものを止める手段です。StreamController はリスナーが去った瞬間を教えてくれます。

var produced = 0;
late final StreamController<int> ctrl;
Timer? timer;
ctrl = StreamController<int>(
  onListen: () {
    print('onListen: start producing');
    timer = Timer.periodic(const Duration(milliseconds: 10), (_) => ctrl.add(++produced));
  },
  onCancel: () {
    print('onCancel: stop producing after $produced events');
    timer?.cancel();
  },
);
final sub = ctrl.stream.listen((v) => v % 5 == 0 ? print('got $v') : null);
await Future.delayed(const Duration(milliseconds: 115));
await sub.cancel();
await Future.delayed(const Duration(milliseconds: 50));
print('after cancel, produced is still $produced, isClosed=${ctrl.isClosed}');

onListen: start producing
got 5
got 10
onCancel: stop producing after 11 events
after cancel, produced is still 11, isClosed=false

sub.cancel() がなければ（Flutter なら dispose() で呼ばなければ）、タイマーはアプリが生きている間ずっと動きます。キャンセルしてもコントローラーは閉じない点も分かります。close() を呼ぶまで isClosed は false です。

Isolate：メイン isolate が止まる様子を測る

isolate は、自分のスレッドとイベントループを持つ独立した Dart のヒープです。isolate 同士はオブジェクトを共有せず、メッセージだけでやり取りします。Isolate.run(f) は isolate を 1 つ起動して f を実行し、結果を返して終了します。「止まる」とはどういうことかを見るため、16ms 周期の「フレーム」タイマーを動かしながら素数を数え、ティック間の最大間隔を記録しました。

import 'dart:async';
import 'dart:isolate';

int countPrimes(int n) {
  var count = 0;
  for (var i = 2; i < n; i++) {
    var prime = true;
    for (var d = 2; d * d <= i; d++) {
      if (i % d == 0) { prime = false; break; }
    }
    if (prime) count++;
  }
  return count;
}

Future<void> measure(String label, Future<int> Function() work) async {
  final sw = Stopwatch()..start();
  var last = 0, maxGap = 0, ticks = 0;
  final t = Timer.periodic(const Duration(milliseconds: 16), (_) {
    final now = sw.elapsedMilliseconds;
    if (now - last > maxGap) maxGap = now - last;
    last = now;
    ticks++;
  });
  await Future.delayed(const Duration(milliseconds: 100)); // settle
  final start = sw.elapsedMilliseconds;
  final ticksBefore = ticks;
  final result = await work();
  final took = sw.elapsedMilliseconds - start;
  final ticksDuring = ticks - ticksBefore;
  await Future.delayed(const Duration(milliseconds: 50));
  t.cancel();
  print('$label: result=$result, work took $took ms, '
      'ticks during work=$ticksDuring (expected ~${took ~/ 16}), '
      'max heartbeat gap=$maxGap ms');
}

Future<void> main() async {
  const n = 3000000;
  await measure('main isolate   ', () async => countPrimes(n));
  await measure('Isolate.run    ', () => Isolate.run(() => countPrimes(n)));
}

main isolate   : result=216816, work took 349 ms, ticks during work=0 (expected ~21), max heartbeat gap=354 ms
Isolate.run    : result=216816, work took 353 ms, ticks during work=22 (expected ~22), max heartbeat gap=18 ms

() async => countPrimes(n) の async は何も変えませんでした。処理中のティックは 0 回、最大間隔は 354ms（他の回は 355、354ms）で、60Hz なら約 21 フレームの欠落です。Isolate.run では処理時間は同じでも、タイマーは動き続けました。ティック 22 回、最大間隔 18ms です。

Flutter の compute(fn, arg) も同じものです。Flutter 3.41.9 の foundation/_isolates_io.dart は単に Isolate.run を呼んでいます。Web には isolate がないので、compute はメインスレッドで関数を実行します。

isolate へデータを送るコスト

isolate はメモリを共有しないので、渡すデータはコピーされます。要素 1,000 万個をワーカーへ送り、長さだけを返してもらいながら 16ms タイマーを動かしました（5 回の中央値。「最大間隔」が 16ms 未満なら遅れたティックはなし）。

教訓は 2 つ。1 つ目、コピーは送信側の isolate で起きます。 List<int> のコピー 28ms はそれだけで 1 フレームより長く、パースを isolate に移しても巨大な List<int> を渡すとまだカクつきうるということです。typed data はバイトが 1 つの塊なので memcpy に近い速さでコピーされます。2 つ目、戻りは安価です。Isolate.run は Isolate.exit で結果のメモリを呼び出し側へそのまま渡します。1,000 万個の List<int> を作るのに、普通なら 81.1ms、isolate で作って受け取ると 89.4ms でした。

よくある間違い：キャプチャしすぎるクロージャ

間違い： 送れないもの（ReceivePort、StreamController、ソケット、Flutter ではよく State オブジェクトそのもの）を一緒に抱えたメソッドの中で、クロージャを書いて Isolate.run を呼ぶ。

import 'dart:isolate';

Future<void> main() async {
  final port = ReceivePort(); // something your State object owns
  final numbers = List<int>.generate(1000, (i) => i);
  void cleanUp() => port.close(); // another closure captures `port`

  final sum = await Isolate.run(() => numbers.reduce((a, b) => a + b));
  print('sum = $sum');
  cleanUp();
}

実際に起きること：

Unhandled exception:
Invalid argument(s): Illegal argument in isolate message: object is unsendable - Library:'dart:isolate' Class: _ReceivePortImpl@1026248 (see restrictions listed at `SendPort.send()` documentation for more information)
 <- Context num_variables: 2 <- Closure: () => int (from ~/demo/capture_trap.dart)
 <- resultPort in Instance of '_RemoteRunner<int>' (from dart:isolate)

クロージャが使ったのは numbers だけですが、同じスコープで作られたクロージャはコンテキストオブジェクトを 1 つ共有しており、その中に port も入っています。実は上の計測コードを書いている最中に偶然踏みました。計算をトップレベル関数 sumAll(numbers) に切り出しても、main の中で Isolate.run(() => sumAll(numbers)) と書くと同じように失敗しました。クロージャがやはり main のスコープで作られるからです。

直し方： 送りたい変数だけを持つ小さな関数の中でクロージャを作ります。

int sumAll(List<int> xs) => xs.reduce((a, b) => a + b);
Future<int> sumInIsolate(List<int> xs) => Isolate.run(() => sumAll(xs));
// main では: final sum = await sumInIsolate(numbers);   -> sum = 499500

Flutter で compute(parseJson, body) にトップレベル関数や static 関数を渡すのが安全なパターンである理由も同じです。

判断表

練習問題

1. 実行せずに出力順を書いてから、動かしてみてください。

import 'dart:async';
void main() {
  Future(() => print('1'));
  Future.value().then((_) => print('2'));
  scheduleMicrotask(() => print('3'));
  print('4');
}

答え

4、2、3、1。同期コード（4）が最初です。次にマイクロタスクが登録順に実行されます。Future.value().then が scheduleMicrotask より先に登録されたので 2、3。Future(() ...) はイベントなので 1 が最後です。

2. ある画面がプロフィール（400ms）とフィード（250ms）を await 2 回で順に読み込んでいます。約 400ms で終わり、値ごとの型を保ち、プロフィールだけ失敗してもフィードは表示するよう書き換えてください。

答え

レコードの .wait を使い、ParallelWaitError を捕まえます。AsyncError は dart:async にあるので、import を忘れるとコンパイラが Error: 'AsyncError' isn't a type. で止まります。

import 'dart:async';
// ...
try {
  final (profile, feed) = await (loadProfile(), loadFeed()).wait;
  show(profile, feed);
} on ParallelWaitError<(Profile?, Feed?), (AsyncError?, AsyncError?)> catch (e) {
  final (profile, feed) = e.values;
  show(profile, feed); // one of them may be null
}

400ms 後に例外を投げるプロフィールと 250ms 後に返るフィードで動かすと、405ms 後に profile=null feed=12 と表示されました。

正直な限界とバージョンメモ

• すべて Dart 3.11.5 でのみ実行しました。レコード、つまり (f1, f2).wait には言語バージョン 3.0 が必要です。// @dart=2.19 コメントを付けると、アナライザーはレコードリテラルを experiment_not_enabled（"This requires the 'records' language feature to be enabled"）で拒否します。Isolate.run はドキュメント上 Dart 2.19 で追加されましたが、古い SDK で直接は確認していません。
• ハートビートのタイマーは Flutter のフレームスケジューラーの代わりであって、スケジューラーそのものではありません。実際のジャンクはラスター時間にも左右され、この実験はそこに触れていません。アプリの数値は実機の profile モードで Flutter DevTools のパフォーマンス画面を見てください。
• コピーのコストはこのマシンとこのサイズでの値です。フィールドの多いオブジェクトは int より要素あたりのコストが大きいので、自分のデータで測ってください。
• Web には isolate がなく、compute はその場で実行されます。

この講座ではコードがどこで動くかを扱いました。第 6 回ではどれだけ速く動くか、つまり JIT と AOT、ウォームアップ、メモリを測ります。

前の講座： 第 4 回：レコード・パターン・sealed クラス · 次の講座： 第 6 回：性能、JIT と AOT、メモリ

関連記事：Swift 第 5 回：並行処理、async と actor · Java の仮想スレッドと並行処理 · Python のスレッド・プロセス・asyncio・GIL · Flutter アプリ 16 本の移行で壊れたもの